Skip to content

Commit b699973

Browse files
fayssalmartanigcpyoshi-code-bot
fayssalmartanigcp
and
yoshi-code-bot
authored
feat: Enable server side flow control by default with the option to turn it off (#426)
* Enable server side flow control by default with the option to turn it off This change enables sending flow control settings automatically to the server. If FlowControlSettings.maxOutstandingElementCount > 0 or FlowControlSettings.maxOutstandingRequestBytes > 0, flow control will be enforced at the server side (in addition to the client side). This behavior is enabled by default and Subscriber.Builder.setUseLegacyFlowControl() method is provided for users who would like to opt-out of this feature in case they encouter issues with server side flow control. * feat: Enable server side flow control by default with the option to turn it off This change enables sending flow control settings automatically to the server. If FlowControlSettings.maxOutstandingElementCount > 0 or FlowControlSettings.maxOutstandingRequestBytes > 0, flow control will be enforced at the server side (in addition to the client side). This behavior is enabled by default and Subscriber.Builder.setUseLegacyFlowControl() method is provided for users who would like to opt-out of this feature in case they encounter issues with server side flow control. * Update google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java Co-authored-by: yoshi-code-bot <[email protected]> * Update google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Co-authored-by: yoshi-code-bot <[email protected]> * Update StreamingSubscriberConnection.java Co-authored-by: yoshi-code-bot <[email protected]>
1 parent c8d2d52 commit b699973

2 files changed

Lines changed: 22 additions & 2 deletions

File tree

‎java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
7373
private final MessageDispatcher messageDispatcher;
7474

7575
private final FlowControlSettings flowControlSettings;
76+
private final boolean useLegacyFlowControl;
7677

7778
private final AtomicLong channelReconnectBackoffMillis =
7879
new AtomicLong(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());
@@ -98,6 +99,7 @@ public StreamingSubscriberConnection(
9899
SubscriberStub stub,
99100
int channelAffinity,
100101
FlowControlSettings flowControlSettings,
102+
boolean useLegacyFlowControl,
101103
FlowController flowController,
102104
ScheduledExecutorService executor,
103105
ScheduledExecutorService systemExecutor,
@@ -119,6 +121,7 @@ public StreamingSubscriberConnection(
119121
systemExecutor,
120122
clock);
121123
this.flowControlSettings = flowControlSettings;
124+
this.useLegacyFlowControl = useLegacyFlowControl;
122125
}
123126

124127
@Override
@@ -217,9 +220,13 @@ private void initialize() {
217220
.setStreamAckDeadlineSeconds(60)
218221
.setClientId(clientId)
219222
.setMaxOutstandingMessages(
220-
valueOrZero(flowControlSettings.getMaxOutstandingElementCount()))
223+
this.useLegacyFlowControl
224+
? 0
225+
: valueOrZero(flowControlSettings.getMaxOutstandingElementCount()))
221226
.setMaxOutstandingBytes(
222-
valueOrZero(flowControlSettings.getMaxOutstandingRequestBytes()))
227+
this.useLegacyFlowControl
228+
? 0
229+
: valueOrZero(flowControlSettings.getMaxOutstandingRequestBytes()))
223230
.build());
224231

225232
/**

‎java-pubsub/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@ public class Subscriber extends AbstractApiService implements SubscriberInterfac
103103

104104
private final String subscriptionName;
105105
private final FlowControlSettings flowControlSettings;
106+
private final boolean useLegacyFlowControl;
106107
private final Duration maxAckExtensionPeriod;
107108
private final Duration maxDurationPerAckExtension;
108109
// The ExecutorProvider used to generate executors for processing messages.
@@ -126,6 +127,7 @@ public class Subscriber extends AbstractApiService implements SubscriberInterfac
126127
private Subscriber(Builder builder) {
127128
receiver = builder.receiver;
128129
flowControlSettings = builder.flowControlSettings;
130+
useLegacyFlowControl = builder.useLegacyFlowControl;
129131
subscriptionName = builder.subscriptionName;
130132

131133
maxAckExtensionPeriod = builder.maxAckExtensionPeriod;
@@ -336,6 +338,7 @@ private void startStreamingConnections() {
336338
subStub,
337339
i,
338340
flowControlSettings,
341+
useLegacyFlowControl,
339342
flowController,
340343
executor,
341344
alarmsExecutor,
@@ -420,6 +423,7 @@ public static final class Builder {
420423
private Duration maxAckExtensionPeriod = DEFAULT_MAX_ACK_EXTENSION_PERIOD;
421424
private Duration maxDurationPerAckExtension = Duration.ofMillis(0);
422425

426+
private boolean useLegacyFlowControl = false;
423427
private FlowControlSettings flowControlSettings =
424428
FlowControlSettings.newBuilder()
425429
.setMaxOutstandingElementCount(1000L)
@@ -504,6 +508,15 @@ public Builder setFlowControlSettings(FlowControlSettings flowControlSettings) {
504508
return this;
505509
}
506510

511+
/**
512+
* Disables enforcing flow control settings at the Cloud PubSub server and uses the less
513+
* accurate method of only enforcing flow control at the client side.
514+
*/
515+
public Builder setUseLegacyFlowControl(boolean value) {
516+
this.useLegacyFlowControl = value;
517+
return this;
518+
}
519+
507520
/**
508521
* Set the maximum period a message ack deadline will be extended. Defaults to one hour.
509522
*

0 commit comments

Comments
 (0)