[Dataflow Streaming] Remove one wait for GetData (#36417)

Harness threads will no longer wait for sending threads directly.
They'll wait on the responseStream and will observe failures when
the responseStream is cancelled.

Reduces context switches and cpu usage under GetData

Author: arunpandianp

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

@@ -502,11 +502,9 @@ private void queueRequestAndWait(QueuedRequest request)
         prevBatch.waitForSendOrFailNotification();
       }
       trySendBatch(batch);
-      // Since the above send may not succeed, we fall through to block on sending or failure.
+      // If the send fails, request.responseStream will be cancelled and
+      // reading responseStream will throw.
     }
-
-    // Wait for this batch to be sent before parsing the response.
-    batch.waitForSendOrFailNotification();
   }
 
   private synchronized void trySendBatch(QueuedBatch batch) throws WindmillStreamShutdownException {

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

@@ -190,13 +190,12 @@ void notifySent() {
       sent.countDown();
     }
 
-    /**
-     * Let waiting for threads know that a failure occurred.
-     *
-     * @implNote Thread safe.
-     */
+    /** Let waiting for threads know that a failure occurred. */
     void notifyFailed() {
       failed = true;
+      for (QueuedRequest request : requests) {
+        request.getResponseStream().cancel();
+      }
       sent.countDown();
     }