---

dpcollins-google · sduskis · commit 27dd7fd466ee · 2019-03-12T18:24:12.000-04:00
yaml --- r: 18787 b: refs/heads/autosynth-pubsub c: 67bf8fc h: refs/heads/master i: 18785: f68eab6 18783: c8d60f3
diff --git a/[refs] b/[refs]
@@ -115,7 +115,7 @@ refs/tags/v0.67.0: 30b56f02092efc6f3c3667650ea8b8825003e0b7
 refs/heads/autosynth-compute: afe19e1678334117f3daf13b968b667857f7c8ca
 refs/heads/autosynth-container: f6384095f50b31bfc8208542a49181e158d3681c
 refs/heads/autosynth-monitoring: 832528c131ac32b115fd7a369701d9e6e4ec62e5
-refs/heads/autosynth-pubsub: 257aa8ea9d3bc3fcb55440cbdd3f550a3400c2be
+refs/heads/autosynth-pubsub: 67bf8fc05b6cdd789f8c5c029de36501670663b2
 refs/heads/autosynth-video-intelligence: e44c9746407fe00fac42e1bb10ac50f493dd37b0
 refs/heads/autosynth-vision: b8e47d76578b5f150ef530072ea7e485e2b02ca0
 refs/heads/spanner: b01127f885b4611bf1852abb0ce481eeb7fcc131
diff --git a/branches/autosynth-pubsub/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/branches/autosynth-pubsub/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java
@@ -99,7 +99,10 @@ public class Subscriber extends AbstractApiService {
   private final String subscriptionName;
   private final FlowControlSettings flowControlSettings;
   private final Duration maxAckExtensionPeriod;
-  private final ScheduledExecutorService executor;
+  // The ExecutorProvider used to generate executors for processing messages.
+  private final ExecutorProvider executorProvider;
+  // An instantiation of the SystemExecutorProvider used for processing acks
+  // and other system actions.
   @Nullable private final ScheduledExecutorService alarmsExecutor;
   private final Distribution ackLatencyDistribution =
       new Distribution(MAX_ACK_DEADLINE_SECONDS + 1);
@@ -132,16 +135,7 @@ private Subscriber(Builder builder) {
 
     this.numPullers = builder.parallelPullCount;
 
-    executor = builder.executorProvider.getExecutor();
-    if (builder.executorProvider.shouldAutoClose()) {
-      closeables.add(
-          new AutoCloseable() {
-            @Override
-            public void close() throws IOException {
-              executor.shutdown();
-            }
-          });
-    }
+    executorProvider = builder.executorProvider;
 
     ExecutorProvider systemExecutorProvider = builder.systemExecutorProvider;
     if (systemExecutorProvider == null) {
@@ -322,6 +316,17 @@ public void run() {
   private void startStreamingConnections() {
     synchronized (streamingSubscriberConnections) {
       for (int i = 0; i < numPullers; i++) {
+        final ScheduledExecutorService executor = executorProvider.getExecutor();
+        if (executorProvider.shouldAutoClose()) {
+          closeables.add(
+              new AutoCloseable() {
+                @Override
+                public void close() {
+                  executor.shutdown();
+                }
+              });
+        }
+
         streamingSubscriberConnections.add(
             new StreamingSubscriberConnection(
                 subscriptionName,
@@ -364,7 +369,7 @@ private void stopAllStreamingConnections() {
   private void startConnections(
       List<? extends ApiService> connections, final ApiService.Listener connectionsListener) {
     for (ApiService subscriber : connections) {
-      subscriber.addListener(connectionsListener, executor);
+      subscriber.addListener(connectionsListener, alarmsExecutor);
       subscriber.startAsync();
     }
     for (ApiService subscriber : connections) {
@@ -398,8 +403,7 @@ public static final class Builder {
 
     static final ExecutorProvider DEFAULT_EXECUTOR_PROVIDER =
         InstantiatingExecutorProvider.newBuilder()
-            .setExecutorThreadCount(
-                THREADS_PER_CHANNEL * Runtime.getRuntime().availableProcessors())
+            .setExecutorThreadCount(THREADS_PER_CHANNEL)
             .build();
 
     String subscriptionName;
@@ -502,7 +506,10 @@ public Builder setMaxAckExtensionPeriod(Duration maxAckExtensionPeriod) {
       return this;
     }
 
-    /** Gives the ability to set a custom executor. */
+    /**
+     * Gives the ability to set a custom executor. {@link ExecutorProvider#getExecutor()} will be
+     * called {@link Builder#parallelPullCount} times.
+     */
     public Builder setExecutorProvider(ExecutorProvider executorProvider) {
       this.executorProvider = Preconditions.checkNotNull(executorProvider);
       return this;
