pubsub: cap stream reconnect time (googleapis#2409)

Previously we exponentially grow stream reconnect backoff time
without bound, so unlucky users will wait forever waiting for
a connection.
This commit caps the reconnect at 10s (the same as polling).

Previously we only reset the polling time after the stream closes
without error. This usually doesn't happen because the streams are
long running and don't close until some error is encountered.
This commit resets the polling time if we receive any response
from the stream.

Fixes googleapis#2406.

Author: pongad

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

@@ -37,6 +37,7 @@
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -48,9 +49,11 @@ final class StreamingSubscriberConnection extends AbstractApiService implements
       Logger.getLogger(StreamingSubscriberConnection.class.getName());
 
   private static final Duration INITIAL_CHANNEL_RECONNECT_BACKOFF = Duration.ofMillis(100);
+  private static final Duration MAX_CHANNEL_RECONNECT_BACKOFF = Duration.ofSeconds(10);
   private static final int MAX_PER_REQUEST_CHANGES = 10000;
 
-  private Duration channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
+  private final AtomicLong channelReconnectBackoffMillis =
+      new AtomicLong(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());
 
   private final SubscriberStub asyncStub;
 
@@ -119,6 +122,7 @@ public void beforeStart(ClientCallStreamObserver<StreamingPullRequest> requestOb
 
     @Override
     public void onNext(StreamingPullResponse response) {
+      channelReconnectBackoffMillis.set(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());
       messageDispatcher.processReceivedMessages(
           response.getReceivedMessagesList(),
           new Runnable() {
@@ -170,7 +174,7 @@ public void onSuccess(@Nullable Void result) {
             if (!isAlive()) {
               return;
             }
-            channelReconnectBackoff = INITIAL_CHANNEL_RECONNECT_BACKOFF;
+            channelReconnectBackoffMillis.set(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());
             // The stream was closed. And any case we want to reopen it to continue receiving
             // messages.
             initialize();
@@ -185,8 +189,11 @@ public void onFailure(Throwable cause) {
             }
             logger.log(Level.WARNING, "Terminated streaming with exception", cause);
             if (StatusUtil.isRetryable(cause)) {
-              long backoffMillis = channelReconnectBackoff.toMillis();
-              channelReconnectBackoff = channelReconnectBackoff.plusMillis(backoffMillis);
+              long backoffMillis = channelReconnectBackoffMillis.get();
+              long newBackoffMillis =
+                  Math.min(backoffMillis * 2, MAX_CHANNEL_RECONNECT_BACKOFF.toMillis());
+              channelReconnectBackoffMillis.set(newBackoffMillis);
+
               executor.schedule(
                   new Runnable() {
                     @Override