Fixing intermittent pubsub test failures

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/PollingSubscriberConnection.java

@@ -16,8 +16,6 @@
 
 package com.google.cloud.pubsub.spi.v1;
 
-import static com.google.cloud.pubsub.spi.v1.StatusUtil.isRetryable;
-
 import com.google.api.gax.core.FlowController;
 import com.google.api.stats.Distribution;
 import com.google.auth.Credentials;
@@ -38,7 +36,6 @@
 import com.google.pubsub.v1.SubscriberGrpc.SubscriberFutureStub;
 import com.google.pubsub.v1.Subscription;
 import io.grpc.Channel;
-import io.grpc.StatusRuntimeException;
 import io.grpc.auth.MoreCallCredentials;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
@@ -167,8 +164,12 @@ public void run() {
 
           @Override
           public void onFailure(Throwable cause) {
-            if (!(cause instanceof StatusRuntimeException)
-                || isRetryable(((StatusRuntimeException) cause).getStatus())) {
+            if (!isAlive()) {
+              // we don't care about subscription failures when we're no longer running.
+              logger.log(Level.FINE, "pull failure after service no longer running", cause);
+              return;
+            }
+            if (StatusUtil.isRetryable(cause)) {
               logger.log(Level.SEVERE, "Failed to pull messages (recoverable): ", cause);
               executor.schedule(
                   new Runnable() {
@@ -183,14 +184,18 @@ public void run() {
                   },
                   backoff.getMillis(),
                   TimeUnit.MILLISECONDS);
-              return;
+            } else {
+              messageDispatcher.stop();
+              notifyFailed(cause);
             }
-            messageDispatcher.stop();
-            notifyFailed(cause);
           }
         });
   }
 
+  private boolean isAlive() {
+    return state() == State.RUNNING || state() == State.STARTING;
+  }
+
   @Override
   public void sendAckOperations(
       List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions) {

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StatusUtil.java

@@ -17,21 +17,31 @@
 package com.google.cloud.pubsub.spi.v1;
 
 import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 
 /** Utilities for handling gRPC {@link Status}. */
 final class StatusUtil {
   private StatusUtil() {
-    // Static class, not instatiable.
+    // Static class, not instantiable.
   }
 
-  public static boolean isRetryable(Status status) {
-    switch (status.getCode()) {
+  public static boolean isRetryable(Throwable error) {
+    if (!(error instanceof StatusRuntimeException)) {
+      return true;
+    }
+    StatusRuntimeException statusRuntimeException = (StatusRuntimeException) error;
+    switch (statusRuntimeException.getStatus().getCode()) {
       case DEADLINE_EXCEEDED:
       case INTERNAL:
       case CANCELLED:
       case RESOURCE_EXHAUSTED:
-      case UNAVAILABLE:
         return true;
+      case UNAVAILABLE:
+        if (statusRuntimeException.getMessage().contains("Server shutdownNow invoked")) {
+          return false;
+        } else {
+          return true;
+        }
       default:
         return false;
     }

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/StreamingSubscriberConnection.java

@@ -16,8 +16,6 @@
 
 package com.google.cloud.pubsub.spi.v1;
 
-import static com.google.cloud.pubsub.spi.v1.StatusUtil.isRetryable;
-
 import com.google.api.gax.core.FlowController;
 import com.google.api.stats.Distribution;
 import com.google.auth.Credentials;
@@ -179,9 +177,13 @@ public void onSuccess(@Nullable Void result) {
           }
 
           @Override
-          public void onFailure(Throwable t) {
-            Status errorStatus = Status.fromThrowable(t);
-            if (isRetryable(errorStatus) && isAlive()) {
+          public void onFailure(Throwable cause) {
+            if (!isAlive()) {
+              // we don't care about subscription failures when we're no longer running.
+              logger.log(Level.FINE, "pull failure after service no longer running", cause);
+              return;
+            }
+            if (StatusUtil.isRetryable(cause)) {
               long backoffMillis = channelReconnectBackoff.getMillis();
               channelReconnectBackoff = channelReconnectBackoff.plus(backoffMillis);
               executor.schedule(
@@ -194,9 +196,7 @@ public void run() {
                   backoffMillis,
                   TimeUnit.MILLISECONDS);
             } else {
-              if (isAlive()) {
-                notifyFailed(t);
-              }
+              notifyFailed(cause);
             }
           }
         },

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/StatusUtilTest.java

@@ -0,0 +1,18 @@
+package com.google.cloud.pubsub.spi.v1;
+
+import com.google.common.truth.Truth;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import org.junit.Test;
+
+public class StatusUtilTest {
+
+  @Test
+  public void testIsRetryable() {
+    Truth.assertThat(StatusUtil.isRetryable(new StatusRuntimeException(Status.UNAVAILABLE)))
+        .isTrue();
+    Truth.assertThat(StatusUtil.isRetryable(new StatusRuntimeException(
+        Status.UNAVAILABLE.withDescription("Server shutdownNow invoked"))))
+        .isFalse();
+  }
+}