---

dpcollins-google · sduskis · commit c5e604e9b9c1 · 2019-03-12T18:24:12.000-04:00
yaml --- r: 21103 b: refs/heads/autosynth-dlp c: 67bf8fc h: refs/heads/master i: 21101: ea9281a 21099: 222337b 21095: 49d315f 21087: e0d6424
diff --git a/[refs] b/[refs]
@@ -103,7 +103,7 @@ refs/tags/v0.60.0: 4cd518d0612329f8a8e53484eef4cd1651e32855
 refs/tags/v0.61.0: e4b526656bb1bf5eefd0ee578b7405147821225e
 refs/tags/v0.62.0: bbede7385d48ba08f487bdd29ec10668ace96396
 refs/heads/0.60.0-alpha: 10939381ffe0b8da32db4fe3087c86e3aa7f3e55
-refs/heads/autosynth-dlp: 257aa8ea9d3bc3fcb55440cbdd3f550a3400c2be
+refs/heads/autosynth-dlp: 67bf8fc05b6cdd789f8c5c029de36501670663b2
 refs/heads/autosynth-logging: ffd2c58e5ab2ff0eb1410bbab4b81682a85a84d2
 refs/heads/dupes: 3478c5d81fd242d0e985656645a679420a2060c2
 refs/tags/v0.63.0: 94f19b71d40f46b36120e7b9d78a1a3d41bfcbd6
diff --git a/branches/autosynth-dlp/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Subscriber.java b/branches/autosynth-dlp/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;
