Skip to content

Commit e92f570

Browse files
michaelbausormichaelbausor
authored
Updates for flowcontrol (#1687)
* Update to accomodate changes to gax * Regenerate with changes to bundling * Remove smoke test, update settings, bump GAX version to 0.3.0
1 parent 5cbbbae commit e92f570

File tree

9 files changed

+101
-106
lines changed

9 files changed

+101
-106
lines changed

‎google-cloud-logging/src/main/java/com/google/cloud/logging/spi/v2/LoggingServiceV2Settings.java‎

Lines changed: 36 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,15 @@
2020
import static com.google.cloud.logging.spi.v2.PagedResponseWrappers.ListMonitoredResourceDescriptorsPagedResponse;
2121

2222
import com.google.api.MonitoredResourceDescriptor;
23+
import com.google.api.gax.bundling.BundlingSettings;
24+
import com.google.api.gax.bundling.RequestBuilder;
25+
import com.google.api.gax.core.FlowControlSettings;
26+
import com.google.api.gax.core.FlowController.LimitExceededBehavior;
2327
import com.google.api.gax.core.GoogleCredentialsProvider;
2428
import com.google.api.gax.core.RetrySettings;
29+
import com.google.api.gax.grpc.BundledRequestIssuer;
2530
import com.google.api.gax.grpc.BundlingCallSettings;
2631
import com.google.api.gax.grpc.BundlingDescriptor;
27-
import com.google.api.gax.grpc.BundlingSettings;
2832
import com.google.api.gax.grpc.CallContext;
2933
import com.google.api.gax.grpc.ChannelProvider;
3034
import com.google.api.gax.grpc.ClientSettings;
@@ -34,7 +38,6 @@
3438
import com.google.api.gax.grpc.PagedCallSettings;
3539
import com.google.api.gax.grpc.PagedListDescriptor;
3640
import com.google.api.gax.grpc.PagedListResponseFactory;
37-
import com.google.api.gax.grpc.RequestIssuer;
3841
import com.google.api.gax.grpc.SimpleCallSettings;
3942
import com.google.api.gax.grpc.UnaryCallSettings;
4043
import com.google.api.gax.grpc.UnaryCallable;
@@ -58,9 +61,7 @@
5861
import com.google.protobuf.ExperimentalApi;
5962
import io.grpc.Status;
6063
import java.io.IOException;
61-
import java.util.ArrayList;
6264
import java.util.Collection;
63-
import java.util.List;
6465
import javax.annotation.Generated;
6566
import org.joda.time.Duration;
6667

@@ -398,33 +399,32 @@ public String getBundlePartitionKey(WriteLogEntriesRequest request) {
398399
}
399400

400401
@Override
401-
public WriteLogEntriesRequest mergeRequests(
402-
Collection<WriteLogEntriesRequest> requests) {
403-
WriteLogEntriesRequest firstRequest = requests.iterator().next();
404-
405-
List<LogEntry> elements = new ArrayList<>();
406-
for (WriteLogEntriesRequest request : requests) {
407-
elements.addAll(request.getEntriesList());
408-
}
409-
410-
WriteLogEntriesRequest bundleRequest =
411-
WriteLogEntriesRequest.newBuilder()
412-
.setLogName(firstRequest.getLogName())
413-
.setResource(firstRequest.getResource())
414-
.putAllLabels(firstRequest.getLabels())
415-
.addAllEntries(elements)
416-
.build();
417-
return bundleRequest;
402+
public RequestBuilder<WriteLogEntriesRequest> getRequestBuilder() {
403+
return new RequestBuilder<WriteLogEntriesRequest>() {
404+
private WriteLogEntriesRequest.Builder builder;
405+
406+
@Override
407+
public void appendRequest(WriteLogEntriesRequest request) {
408+
if (builder == null) {
409+
builder = request.toBuilder();
410+
} else {
411+
builder.addAllEntries(request.getEntriesList());
412+
}
413+
}
414+
415+
@Override
416+
public WriteLogEntriesRequest build() {
417+
return builder.build();
418+
}
419+
};
418420
}
419421

420422
@Override
421423
public void splitResponse(
422424
WriteLogEntriesResponse bundleResponse,
423-
Collection<? extends RequestIssuer<WriteLogEntriesRequest, WriteLogEntriesResponse>>
424-
bundle) {
425+
Collection<? extends BundledRequestIssuer<WriteLogEntriesResponse>> bundle) {
425426
int bundleMessageIndex = 0;
426-
for (RequestIssuer<WriteLogEntriesRequest, WriteLogEntriesResponse> responder :
427-
bundle) {
427+
for (BundledRequestIssuer<WriteLogEntriesResponse> responder : bundle) {
428428
WriteLogEntriesResponse response = WriteLogEntriesResponse.newBuilder().build();
429429
responder.setResponse(response);
430430
}
@@ -433,10 +433,8 @@ public void splitResponse(
433433
@Override
434434
public void splitException(
435435
Throwable throwable,
436-
Collection<? extends RequestIssuer<WriteLogEntriesRequest, WriteLogEntriesResponse>>
437-
bundle) {
438-
for (RequestIssuer<WriteLogEntriesRequest, WriteLogEntriesResponse> responder :
439-
bundle) {
436+
Collection<? extends BundledRequestIssuer<WriteLogEntriesResponse>> bundle) {
437+
for (BundledRequestIssuer<WriteLogEntriesResponse> responder : bundle) {
440438
responder.setException(throwable);
441439
}
442440
}
@@ -556,9 +554,15 @@ private static Builder createDefault() {
556554
builder
557555
.writeLogEntriesSettings()
558556
.getBundlingSettingsBuilder()
559-
.setElementCountThreshold(100)
560-
.setRequestByteThreshold(1024)
561-
.setDelayThreshold(Duration.millis(10));
557+
.setElementCountThreshold(1000)
558+
.setRequestByteThreshold(1048576)
559+
.setDelayThreshold(Duration.millis(50))
560+
.setFlowControlSettings(
561+
FlowControlSettings.newBuilder()
562+
.setMaxOutstandingElementCount(100000)
563+
.setMaxOutstandingRequestBytes(10485760)
564+
.setLimitExceededBehavior(LimitExceededBehavior.ThrowException)
565+
.build());
562566
builder
563567
.writeLogEntriesSettings()
564568
.setRetryableCodes(RETRYABLE_CODE_DEFINITIONS.get("non_idempotent"))

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616

1717
package com.google.cloud.pubsub.spi.v1;
1818

19-
import com.google.api.gax.grpc.FlowController;
19+
import com.google.api.gax.core.FlowController;
2020
import com.google.api.stats.Distribution;
2121
import com.google.cloud.Clock;
2222
import com.google.common.annotations.VisibleForTesting;

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818

1919
import static com.google.cloud.pubsub.spi.v1.StatusUtil.isRetryable;
2020

21-
import com.google.api.gax.grpc.FlowController;
21+
import com.google.api.gax.core.FlowController;
2222
import com.google.api.stats.Distribution;
2323
import com.google.auth.Credentials;
2424
import com.google.cloud.Clock;

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

Lines changed: 18 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -16,16 +16,16 @@
1616

1717
package com.google.cloud.pubsub.spi.v1;
1818

19+
import com.google.api.gax.bundling.BundlingSettings;
1920
import com.google.api.gax.core.ApiFuture;
2021
import com.google.api.gax.core.ApiFutureCallback;
2122
import com.google.api.gax.core.ApiFutures;
23+
import com.google.api.gax.core.FlowControlSettings;
24+
import com.google.api.gax.core.FlowController;
2225
import com.google.api.gax.core.Function;
2326
import com.google.api.gax.core.RetrySettings;
24-
import com.google.api.gax.grpc.BundlingSettings;
2527
import com.google.api.gax.grpc.ChannelProvider;
2628
import com.google.api.gax.grpc.ExecutorProvider;
27-
import com.google.api.gax.grpc.FlowControlSettings;
28-
import com.google.api.gax.grpc.FlowController;
2929
import com.google.api.gax.grpc.InstantiatingExecutorProvider;
3030
import com.google.auth.oauth2.GoogleCredentials;
3131
import com.google.common.annotations.VisibleForTesting;
@@ -88,7 +88,6 @@ public class Publisher {
8888
private final LongRandom longRandom;
8989

9090
private final FlowControlSettings flowControlSettings;
91-
private final boolean failOnFlowControlLimits;
9291

9392
private final Lock messagesBundleLock;
9493
private List<OutstandingPublish> messagesBundle;
@@ -125,8 +124,7 @@ private Publisher(Builder builder) throws IOException {
125124
this.longRandom = builder.longRandom;
126125

127126
flowControlSettings = builder.flowControlSettings;
128-
failOnFlowControlLimits = builder.failOnFlowControlLimits;
129-
this.flowController = new FlowController(flowControlSettings, failOnFlowControlLimits);
127+
this.flowController = new FlowController(flowControlSettings);
130128

131129
messagesBundle = new LinkedList<>();
132130
messagesBundleLock = new ReentrantLock();
@@ -173,12 +171,14 @@ public TopicName getTopicName() {
173171
* Schedules the publishing of a message. The publishing of the message may occur immediately or
174172
* be delayed based on the publisher bundling options.
175173
*
176-
* <p>Depending on chosen flow control {@link #failOnFlowControlLimits option}, the returned
177-
* future might immediately fail with a {@link com.google.api.gax.grpc.FlowController.FlowControlException}
178-
* or block the current thread until there are more resources available to publish.
174+
* <p>Depending on chosen flow control {@link FlowControlSettings#getLimitExceededBehavior
175+
* option}, the returned future might immediately fail with a {@link
176+
* FlowController.FlowControlException} or block the current thread until there are more resources
177+
* available to publish.
179178
*
180179
* <p>Example of publishing a message.
181-
* <pre> {@code
180+
*
181+
* <pre>{@code
182182
* String message = "my_message";
183183
* ByteString data = ByteString.copyFromUtf8(message);
184184
* PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();
@@ -473,26 +473,19 @@ private long getMaxBundleBytes() {
473473
}
474474

475475
/**
476-
* The bundling settings configured on this {@code Publisher}. See {@link
477-
* #failOnFlowControlLimits()}.
476+
* The bundling settings configured on this {@code Publisher}, including whether to block publish
477+
* calls when reaching flow control limits.
478+
*
479+
* <p>If {@link FlowControlSettings#getLimitExceededBehavior()} is set to {@link
480+
* FlowController.LimitExceededBehavior#ThrowException}, a publish call will fail with either
481+
* {@link FlowController.MaxOutstandingRequestBytesReachedException} or {@link
482+
* FlowController.MaxOutstandingElementCountReachedException}, as appropriate, when flow control
483+
* limits are reached.
478484
*/
479485
public FlowControlSettings getFlowControlSettings() {
480486
return flowControlSettings;
481487
}
482488

483-
/**
484-
* Whether to block publish calls when reaching flow control limits (see {@link
485-
* #getFlowControlSettings()}).
486-
*
487-
* <p>If set to false, a publish call will fail with either {@link
488-
* com.google.api.gax.grpc.FlowController.MaxOutstandingRequestBytesReachedException} or {@link
489-
* com.google.api.gax.grpc.FlowController.MaxOutstandingElementCountReachedException}, as
490-
* appropriate, when flow contro limits are reached.
491-
*/
492-
public boolean failOnFlowControlLimits() {
493-
return failOnFlowControlLimits;
494-
}
495-
496489
/**
497490
* Schedules immediate publishing of any outstanding messages and waits until all are processed.
498491
*
@@ -619,7 +612,6 @@ public long nextLong(long least, long bound) {
619612

620613
// Client-side flow control options
621614
FlowControlSettings flowControlSettings = FlowControlSettings.getDefaultInstance();
622-
boolean failOnFlowControlLimits;
623615

624616
RetrySettings retrySettings = DEFAULT_RETRY_SETTINGS;
625617
LongRandom longRandom = DEFAULT_LONG_RANDOM;
@@ -665,20 +657,6 @@ public Builder setFlowControlSettings(FlowControlSettings flowControlSettings) {
665657
return this;
666658
}
667659

668-
/**
669-
* Whether to fail publish when reaching any of the flow control limits, with either a {@link
670-
* com.google.api.gax.grpc.FlowController.MaxOutstandingRequestBytesReachedException} or {@link
671-
* com.google.api.gax.grpc.FlowController.MaxOutstandingElementCountReachedException} as
672-
* appropriate.
673-
*
674-
* <p>If set to false, then publish operations will block the current thread until the
675-
* outstanding requests go under the limits.
676-
*/
677-
public Builder setFailOnFlowControlLimits(boolean fail) {
678-
failOnFlowControlLimits = fail;
679-
return this;
680-
}
681-
682660
/** Configures the Publisher's retry parameters. */
683661
public Builder setRetrySettings(RetrySettings retrySettings) {
684662
Preconditions.checkArgument(

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

Lines changed: 31 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,15 @@
1818
import static com.google.cloud.pubsub.spi.v1.PagedResponseWrappers.ListTopicSubscriptionsPagedResponse;
1919
import static com.google.cloud.pubsub.spi.v1.PagedResponseWrappers.ListTopicsPagedResponse;
2020

21+
import com.google.api.gax.bundling.BundlingSettings;
22+
import com.google.api.gax.bundling.RequestBuilder;
23+
import com.google.api.gax.core.FlowControlSettings;
24+
import com.google.api.gax.core.FlowController.LimitExceededBehavior;
2125
import com.google.api.gax.core.GoogleCredentialsProvider;
2226
import com.google.api.gax.core.RetrySettings;
27+
import com.google.api.gax.grpc.BundledRequestIssuer;
2328
import com.google.api.gax.grpc.BundlingCallSettings;
2429
import com.google.api.gax.grpc.BundlingDescriptor;
25-
import com.google.api.gax.grpc.BundlingSettings;
2630
import com.google.api.gax.grpc.CallContext;
2731
import com.google.api.gax.grpc.ChannelProvider;
2832
import com.google.api.gax.grpc.ClientSettings;
@@ -32,7 +36,6 @@
3236
import com.google.api.gax.grpc.PagedCallSettings;
3337
import com.google.api.gax.grpc.PagedListDescriptor;
3438
import com.google.api.gax.grpc.PagedListResponseFactory;
35-
import com.google.api.gax.grpc.RequestIssuer;
3639
import com.google.api.gax.grpc.SimpleCallSettings;
3740
import com.google.api.gax.grpc.UnaryCallSettings;
3841
import com.google.api.gax.grpc.UnaryCallable;
@@ -58,7 +61,6 @@
5861
import com.google.pubsub.v1.PublishRequest;
5962
import com.google.pubsub.v1.PublishResponse;
6063
import com.google.pubsub.v1.PublisherGrpc;
61-
import com.google.pubsub.v1.PubsubMessage;
6264
import com.google.pubsub.v1.Topic;
6365
import io.grpc.Status;
6466
import java.io.IOException;
@@ -363,30 +365,34 @@ public String getBundlePartitionKey(PublishRequest request) {
363365
}
364366

365367
@Override
366-
public PublishRequest mergeRequests(Collection<PublishRequest> requests) {
367-
PublishRequest firstRequest = requests.iterator().next();
368+
public RequestBuilder<PublishRequest> getRequestBuilder() {
369+
return new RequestBuilder<PublishRequest>() {
370+
private PublishRequest.Builder builder;
368371

369-
List<PubsubMessage> elements = new ArrayList<>();
370-
for (PublishRequest request : requests) {
371-
elements.addAll(request.getMessagesList());
372-
}
372+
@Override
373+
public void appendRequest(PublishRequest request) {
374+
if (builder == null) {
375+
builder = request.toBuilder();
376+
} else {
377+
builder.addAllMessages(request.getMessagesList());
378+
}
379+
}
373380

374-
PublishRequest bundleRequest =
375-
PublishRequest.newBuilder()
376-
.setTopic(firstRequest.getTopic())
377-
.addAllMessages(elements)
378-
.build();
379-
return bundleRequest;
381+
@Override
382+
public PublishRequest build() {
383+
return builder.build();
384+
}
385+
};
380386
}
381387

382388
@Override
383389
public void splitResponse(
384390
PublishResponse bundleResponse,
385-
Collection<? extends RequestIssuer<PublishRequest, PublishResponse>> bundle) {
391+
Collection<? extends BundledRequestIssuer<PublishResponse>> bundle) {
386392
int bundleMessageIndex = 0;
387-
for (RequestIssuer<PublishRequest, PublishResponse> responder : bundle) {
393+
for (BundledRequestIssuer<PublishResponse> responder : bundle) {
388394
List<String> subresponseElements = new ArrayList<>();
389-
int subresponseCount = responder.getRequest().getMessagesCount();
395+
long subresponseCount = responder.getMessageCount();
390396
for (int i = 0; i < subresponseCount; i++) {
391397
subresponseElements.add(bundleResponse.getMessageIds(bundleMessageIndex));
392398
bundleMessageIndex += 1;
@@ -400,8 +406,8 @@ public void splitResponse(
400406
@Override
401407
public void splitException(
402408
Throwable throwable,
403-
Collection<? extends RequestIssuer<PublishRequest, PublishResponse>> bundle) {
404-
for (RequestIssuer<PublishRequest, PublishResponse> responder : bundle) {
409+
Collection<? extends BundledRequestIssuer<PublishResponse>> bundle) {
410+
for (BundledRequestIssuer<PublishResponse> responder : bundle) {
405411
responder.setException(throwable);
406412
}
407413
}
@@ -539,7 +545,11 @@ private static Builder createDefault() {
539545
.getBundlingSettingsBuilder()
540546
.setElementCountThreshold(10)
541547
.setRequestByteThreshold(1024)
542-
.setDelayThreshold(Duration.millis(10));
548+
.setDelayThreshold(Duration.millis(10))
549+
.setFlowControlSettings(
550+
FlowControlSettings.newBuilder()
551+
.setLimitExceededBehavior(LimitExceededBehavior.Ignore)
552+
.build());
543553
builder
544554
.publishSettings()
545555
.setRetryableCodes(RETRYABLE_CODE_DEFINITIONS.get("one_plus_delivery"))

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818

1919
import static com.google.cloud.pubsub.spi.v1.StatusUtil.isRetryable;
2020

21-
import com.google.api.gax.grpc.FlowController;
21+
import com.google.api.gax.core.FlowController;
2222
import com.google.api.stats.Distribution;
2323
import com.google.auth.Credentials;
2424
import com.google.cloud.Clock;

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,9 @@
1616

1717
package com.google.cloud.pubsub.spi.v1;
1818

19+
import com.google.api.gax.core.FlowControlSettings;
20+
import com.google.api.gax.core.FlowController;
1921
import com.google.api.gax.grpc.ExecutorProvider;
20-
import com.google.api.gax.grpc.FlowControlSettings;
21-
import com.google.api.gax.grpc.FlowController;
2222
import com.google.api.gax.grpc.InstantiatingExecutorProvider;
2323
import com.google.api.stats.Distribution;
2424
import com.google.auth.Credentials;
@@ -283,7 +283,7 @@ private SubscriberImpl(Builder builder) throws IOException {
283283
Ints.saturatedCast(ackExpirationPadding.getStandardSeconds()));
284284
clock = builder.clock.isPresent() ? builder.clock.get() : Clock.defaultClock();
285285

286-
flowController = new FlowController(builder.flowControlSettings, false);
286+
flowController = new FlowController(builder.flowControlSettings);
287287

288288
executor = builder.executorProvider.getExecutor();
289289
if (builder.executorProvider.shouldAutoClose()) {

0 commit comments

Comments
 (0)