---

dpcollins-google · sduskis · commit 5ce3af730a05 · 2019-02-28T14:25:24.000-05:00
yaml --- r: 22283 b: refs/heads/autosynth-asset c: f5fae7e h: refs/heads/master i: 22281: 5f26d4f 22279: 5a9a515
diff --git a/[refs] b/[refs]
@@ -120,7 +120,7 @@ refs/heads/autosynth-vision: 3866c747a5e87b5dfd530d7134907a7ed1fb16de
 refs/heads/spanner: b01127f885b4611bf1852abb0ce481eeb7fcc131
 refs/tags/v0.68.0: 9cc799fcf68c82ab431d425fefa58ef615ce8e5b
 refs/tags/v0.69.0: 78f67a29e8b9c46ba01de566a2eae0fd1c03edea
-refs/heads/autosynth-asset: b24085700f045f878818380e9cbf6d056a710a33
+refs/heads/autosynth-asset: f5fae7efce06272dffd4f76b9cfa2ae13dee3c68
 refs/heads/autosynth-automl: 2a8b018cf05811fd472e5d1053e67a12ceec0b64
 refs/heads/autosynth-bigquerydatatransfer: 564833a85642d4194adc025f021e10e723154246
 refs/heads/autosynth-bigquerystorage: a75c34ed7a11741669121be69a7021a00f1133ce
diff --git a/branches/autosynth-asset/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/branches/autosynth-asset/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)
