@@ -103,12 +103,7 @@ private StreamingSubscriberConnection(Builder builder) {
103103 // We need to set the default stream ack deadline on the initial request, this will be
104104 // updated by modack requests in the message dispatcher
105105 if (builder .maxDurationPerAckExtensionDefaultUsed ) {
106- // If the default is used, check if exactly once is enabled and set appropriately
107- if (builder .exactlyOnceDeliveryEnabled ) {
108- inititalStreamAckDeadline = Subscriber .STREAM_ACK_DEADLINE_EXACTLY_ONCE_DELIVERY_DEFAULT ;
109- } else {
110- inititalStreamAckDeadline = Subscriber .STREAM_ACK_DEADLINE_DEFAULT ;
111- }
106+ inititalStreamAckDeadline = Subscriber .STREAM_ACK_DEADLINE_DEFAULT ;
112107 } else if (builder .maxDurationPerAckExtension .compareTo (Subscriber .MIN_STREAM_ACK_DEADLINE )
113108 < 0 ) {
114109 // We will not be able to extend more than the default minimum
@@ -123,7 +118,6 @@ private StreamingSubscriberConnection(Builder builder) {
123118
124119 subscriberStub = builder .subscriberStub ;
125120 channelAffinity = builder .channelAffinity ;
126- exactlyOnceDeliveryEnabled .set (builder .exactlyOnceDeliveryEnabled );
127121
128122 MessageDispatcher .Builder messageDispatcherBuilder ;
129123 if (builder .receiver != null ) {
@@ -143,7 +137,6 @@ private StreamingSubscriberConnection(Builder builder) {
143137 .setMaxDurationPerAckExtensionDefaultUsed (builder .maxDurationPerAckExtensionDefaultUsed )
144138 .setAckLatencyDistribution (builder .ackLatencyDistribution )
145139 .setFlowController (builder .flowController )
146- .setEnableExactlyOnceDelivery (builder .exactlyOnceDeliveryEnabled )
147140 .setExecutor (builder .executor )
148141 .setSystemExecutor (builder .systemExecutor )
149142 .setApiClock (builder .clock )
@@ -159,7 +152,7 @@ public StreamingSubscriberConnection setExactlyOnceDeliveryEnabled(
159152 return this ;
160153 }
161154
162- public boolean isExactlyOnceDeliveryEnabled () {
155+ public boolean getExactlyOnceDeliveryEnabled () {
163156 return exactlyOnceDeliveryEnabled .get ();
164157 }
165158
@@ -221,7 +214,7 @@ public void onResponse(StreamingPullResponse response) {
221214 response .getSubscriptionProperties ().getExactlyOnceDeliveryEnabled ();
222215
223216 setExactlyOnceDeliveryEnabled (exactlyOnceDeliveryEnabledResponse );
224- messageDispatcher .setEnableExactlyOnceDelivery (exactlyOnceDeliveryEnabledResponse );
217+ messageDispatcher .setExactlyOnceDeliveryEnabled (exactlyOnceDeliveryEnabledResponse );
225218 messageDispatcher .processReceivedMessages (response .getReceivedMessagesList ());
226219
227220 // Only request more if we're not shutdown.
@@ -370,7 +363,7 @@ public void setResponseOutstandingMessages(AckResponse ackResponse) {
370363 private void setFailureFutureOutstandingMessages (Throwable t ) {
371364 AckResponse ackResponse ;
372365
373- if (isExactlyOnceDeliveryEnabled ()) {
366+ if (getExactlyOnceDeliveryEnabled ()) {
374367 if (!(t instanceof ApiException )) {
375368 ackResponse = AckResponse .OTHER ;
376369 }
@@ -518,7 +511,7 @@ public void onFailure(Throwable t) {
518511 // Remove from our pending operations
519512 ackOperationsWaiter .incrementPendingCount (-1 );
520513
521- if (!isExactlyOnceDeliveryEnabled ()) {
514+ if (!getExactlyOnceDeliveryEnabled ()) {
522515 Level level = isAlive () ? Level .WARNING : Level .FINER ;
523516 logger .log (level , "failed to send operations" , t );
524517 return ;
@@ -609,7 +602,6 @@ public static final class Builder {
609602 private int channelAffinity ;
610603 private FlowController flowController ;
611604 private FlowControlSettings flowControlSettings ;
612- private boolean exactlyOnceDeliveryEnabled ;
613605 private boolean useLegacyFlowControl ;
614606 private ScheduledExecutorService executor ;
615607 private ScheduledExecutorService systemExecutor ;
@@ -690,11 +682,6 @@ public Builder setUseLegacyFlowControl(boolean useLegacyFlowControl) {
690682 return this ;
691683 }
692684
693- public Builder setExactlyOnceDeliveryEnabled (boolean exactlyOnceDeliveryEnabled ) {
694- this .exactlyOnceDeliveryEnabled = exactlyOnceDeliveryEnabled ;
695- return this ;
696- }
697-
698685 public Builder setExecutor (ScheduledExecutorService executor ) {
699686 this .executor = executor ;
700687 return this ;
0 commit comments