pubsub: start subscriber connections quickly again

pongad · pongad · commit 07f1c28edb1d · 2017-06-01T13:46:37.000+10:00
This PR undoes part of #2055. Refreshing credentials blocks thread for a period of time. Previously, each channel has its own credentials, so starting many channels spams executor with many blocking tasks. but the downside is we take a long time to fully start: about 2 minutes on my machine. Now, all connections share the same credentials, so refreshing only happens once. We can start quickly again. Also rename setLeaseAlarmsExecutorProvider, since the executor is now also used for polling.
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java
@@ -335,9 +335,10 @@ private void startConnections(
       List<? extends ApiService> connections, final ApiService.Listener connectionsListener) {
     for (ApiService subscriber : connections) {
       subscriber.addListener(connectionsListener, executor);
-      // Starting each connection submits a blocking task to the executor.
-      // We start connections one at a time to avoid swamping executor with blocking tasks.
-      subscriber.startAsync().awaitRunning();
+      subscriber.startAsync();
+    }
+    for (ApiService subscriber : connections) {
+      subscriber.awaitRunning();
     }
   }
 
@@ -468,10 +469,10 @@ public Builder setCredentialsProvider(CredentialsProvider credentialsProvider) {
     }
 
     /**
-     * Gives the ability to set a custom executor for managing lease extensions. If none is provided
-     * a shared one will be used by all {@link Subscriber} instances.
+     * Gives the ability to set a custom executor for polling and managing lease extensions. If none
+     * is provided a shared one will be used by all {@link Subscriber} instances.
      */
-    public Builder setLeaseAlarmsExecutorProvider(ExecutorProvider executorProvider) {
+    public Builder setSystemExecutorProvider(ExecutorProvider executorProvider) {
       this.systemExecutorProvider = Preconditions.checkNotNull(executorProvider);
       return this;
     }
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberTest.java
@@ -536,7 +536,7 @@ private void sendMessages(Iterable<String> ackIds) throws InterruptedException {
   private Builder getTestSubscriberBuilder(MessageReceiver receiver) {
     return Subscriber.defaultBuilder(TEST_SUBSCRIPTION, receiver)
         .setExecutorProvider(FixedExecutorProvider.create(fakeExecutor))
-        .setLeaseAlarmsExecutorProvider(FixedExecutorProvider.create(fakeExecutor))
+        .setSystemExecutorProvider(FixedExecutorProvider.create(fakeExecutor))
         .setChannelProvider(FixedChannelProvider.create(testChannel))
         .setCredentialsProvider(PublisherImplTest.NO_CREDENTIALS_PROVIDER)
         .setClock(fakeExecutor.getClock());
