[Dataflow Streaming] Change GrpcGetDataStream to backoff requests that have been cancelled. (#36475)

Author: scwhittle

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStream.java

@@ -80,7 +80,7 @@ final class GrpcGetDataStream
 
   static final FluentBackoff BACK_OFF_FACTORY =
       FluentBackoff.DEFAULT
-          .withInitialBackoff(Duration.millis(10))
+          .withInitialBackoff(Duration.millis(1))
           .withMaxBackoff(Duration.standardSeconds(10));
 
   /**
@@ -439,23 +439,24 @@ private <ResponseT> ResponseT issueRequest(QueuedRequest request, ParseFn<Respon
       try {
         queueRequestAndWait(request);
         return parseFn.parse(request.getResponseStream());
-      } catch (AppendableInputStream.InvalidInputStreamStateException | CancellationException e) {
+      } catch (CancellationException e) {
         throwIfShutdown(request, e);
-        if (!(e instanceof CancellationException)) {
-          throw e;
-        }
+      } catch (AppendableInputStream.InvalidInputStreamStateException e) {
+        throwIfShutdown(request, e);
+        throw e;
       } catch (IOException e) {
         LOG.error("Parsing GetData response failed: ", e);
-        try {
-          BackOffUtils.next(Sleeper.DEFAULT, backoff);
-        } catch (InterruptedException ie) {
-          Thread.currentThread().interrupt();
-        }
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         throwIfShutdown(request, e);
         throw new RuntimeException(e);
       }
+      // In all cases we are going to retry, perform some backoff
+      try {
+        BackOffUtils.next(Sleeper.DEFAULT, backoff);
+      } catch (InterruptedException ie) {
+        Thread.currentThread().interrupt();
+      }
     }
   }