pubsub: use one thread to start/stop connections

pongad · pongad · commit 4928775859b2 · 2017-05-10T18:19:50.000+10:00
Previously we submit a bunch of tasks which is unnecessary and deadlock-prone. This does not solve the one-thread-deadlock problem, but it changes the symptom. In my testing, Subscriber now successfully pulls one batch of messages before failing due to DEADLINE_EXCEEDED error. This is obviously undesirable but at least it doesn't deadlock anymore. This is my best attempt at reconstructing the events: 1. A bunch of tasks get registered to [GetSubscription](https://github.com/GoogleCloudPlatform/google-cloud-java/blob/9623994f58199c79ee5b9f99ad0ff6d7fb69bd84/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java#L105). 2. At least one of these tasks succeed and continue to [pull messsages](https://github.com/GoogleCloudPlatform/google-cloud-java/blob/9623994f58199c79ee5b9f99ad0ff6d7fb69bd84/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java#L132). 3. Something somewhere register a very long-running task (~30sec on my machine) to the pool. Stack trace (obtained by `jstack`) revealed that the pool is busy doing something, but I don't know what it is yet. 4. A message is pulled and the task of printing it to the screen is registered to the pool. It cannot run yet because the long-running task is using the only thread available. 5. At least one of tasks that called GetSubscription timed out. 6. Long running task timed out or completed. 7. Message finally printed to the screen. 8. Because time out on GetSubscription is not considered retryable, the Subscriber fails. It only failed after 30sec because the task to fail it is also blocked behind the long-running task. Updates #1827 and #2041.
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
@@ -387,25 +387,12 @@ private void stopAllPollingConnections() {
 
   private void startConnections(
       List<? extends ApiService> connections, final ApiService.Listener connectionsListener) {
-    final CountDownLatch subscribersStarting = new CountDownLatch(numChannels);
-    for (final ApiService subscriber : connections) {
-      executor.submit(
-          new Runnable() {
-            @Override
-            public void run() {
-              subscriber.addListener(connectionsListener, executor);
-              try {
-                subscriber.startAsync().awaitRunning();
-              } finally {
-                subscribersStarting.countDown();
-              }
-            }
-          });
+    for (ApiService subscriber : connections) {
+      subscriber.addListener(connectionsListener, executor);
+      subscriber.startAsync();
     }
-    try {
-      subscribersStarting.await();
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
+    for (ApiService subscriber : connections) {
+      subscriber.awaitRunning();
     }
   }
 
@@ -415,26 +402,16 @@ private void stopConnections(List<? extends ApiService> connections) {
       liveConnections = new ArrayList<ApiService>(connections);
       connections.clear();
     }
-    final CountDownLatch connectionsStopping = new CountDownLatch(liveConnections.size());
-    for (final ApiService subscriberConnection : liveConnections) {
-      executor.submit(
-          new Runnable() {
-            @Override
-            public void run() {
-              try {
-                subscriberConnection.stopAsync().awaitTerminated();
-              } catch (IllegalStateException ignored) {
-                // It is expected for some connections to be already in state failed so stop will
-                // throw this expection.
-              }
-              connectionsStopping.countDown();
-            }
-          });
+    for (ApiService subscriber : liveConnections) {
+      subscriber.stopAsync();
     }
-    try {
-      connectionsStopping.await();
-    } catch (InterruptedException e) {
-      throw new IllegalStateException(e);
+    for (ApiService subscriber : liveConnections) {
+      try {
+        subscriber.awaitTerminated();
+      } catch (IllegalStateException e) {
+        // It is expected for some connections to be already in state failed so stop will
+        // throw this expection.
+      }
     }
   }
 
