---

dpcollins-google · sduskis · commit f049ea4a1cd1 · 2019-02-28T14:25:24.000-05:00
yaml --- r: 25979 b: refs/heads/pubsub-ordering-keys c: f5fae7e h: refs/heads/master i: 25977: d1db052 25975: a2c0d7f
diff --git a/[refs] b/[refs]
@@ -155,7 +155,7 @@ refs/tags/v0.72.0: a7703f2593ba312c0b2dde6fdfd4f5c764bb55ac
 refs/tags/v0.73.0: 21241ea8be9439cc5764c4944cdce21d34ce4f9e
 refs/tags/v0.74.0: 9d1f733dbbf790de7b494418523b69c4a9a57638
 refs/heads/ignoretest: 23c412ae07af3d0ab1caa2d44d5bc5c0ccb8b31d
-refs/heads/pubsub-ordering-keys: b24085700f045f878818380e9cbf6d056a710a33
+refs/heads/pubsub-ordering-keys: f5fae7efce06272dffd4f76b9cfa2ae13dee3c68
 "refs/heads/update_mvn_badge": ae2d773814db0f71197ccf5a8612ee1d8056f8de
 refs/tags/v0.75.0: c3673089ae09a897c1b4cf7dfe167fe4f8ab32fb
 refs/tags/v0.76.0: 395b016826d3ddf9cb8b34919636df15a4dbd032
diff --git a/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/branches/pubsub-ordering-keys/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
@@ -45,6 +45,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -93,9 +94,6 @@ public class Subscriber extends AbstractApiService {
   private static final Duration UNARY_TIMEOUT = Duration.ofSeconds(60);
   private static final Duration ACK_EXPIRATION_PADDING = Duration.ofSeconds(5);
 
-  private static final ScheduledExecutorService SHARED_SYSTEM_EXECUTOR =
-      InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(6).build().getExecutor();
-
   private static final Logger logger = Logger.getLogger(Subscriber.class.getName());
 
   private final String subscriptionName;
@@ -132,6 +130,8 @@ private Subscriber(Builder builder) {
                 .setLimitExceededBehavior(LimitExceededBehavior.ThrowException)
                 .build());
 
+    this.numPullers = builder.parallelPullCount;
+
     executor = builder.executorProvider.getExecutor();
     if (builder.executorProvider.shouldAutoClose()) {
       closeables.add(
@@ -142,8 +142,16 @@ public void close() throws IOException {
             }
           });
     }
-    alarmsExecutor = builder.systemExecutorProvider.getExecutor();
-    if (builder.systemExecutorProvider.shouldAutoClose()) {
+
+    ExecutorProvider systemExecutorProvider = builder.systemExecutorProvider;
+    if (systemExecutorProvider == null) {
+      systemExecutorProvider =
+          FixedExecutorProvider.create(
+              Executors.newScheduledThreadPool(Math.max(6, 2 * numPullers)));
+    }
+
+    alarmsExecutor = systemExecutorProvider.getExecutor();
+    if (systemExecutorProvider.shouldAutoClose()) {
       closeables.add(
           new AutoCloseable() {
             @Override
@@ -153,7 +161,6 @@ public void close() throws IOException {
           });
     }
 
-    this.numPullers = builder.parallelPullCount;
     TransportChannelProvider channelProvider = builder.channelProvider;
     if (channelProvider.acceptsPoolSize()) {
       channelProvider = channelProvider.withPoolSize(numPullers);
@@ -162,7 +169,7 @@ public void close() throws IOException {
     try {
       this.subStubSettings =
           SubscriberStubSettings.newBuilder()
-              .setExecutorProvider(FixedExecutorProvider.create(alarmsExecutor))
+              .setExecutorProvider(systemExecutorProvider)
               .setCredentialsProvider(builder.credentialsProvider)
               .setTransportChannelProvider(channelProvider)
               .setHeaderProvider(builder.headerProvider)
@@ -404,7 +411,7 @@ public static final class Builder {
         FlowControlSettings.newBuilder().setMaxOutstandingElementCount(1000L).build();
 
     ExecutorProvider executorProvider = DEFAULT_EXECUTOR_PROVIDER;
-    ExecutorProvider systemExecutorProvider = FixedExecutorProvider.create(SHARED_SYSTEM_EXECUTOR);
+    ExecutorProvider systemExecutorProvider = null;
     TransportChannelProvider channelProvider =
         SubscriptionAdminSettings.defaultGrpcTransportProviderBuilder()
             .setMaxInboundMessageSize(MAX_INBOUND_MESSAGE_SIZE)
