Re: [PR] do not block other threads from health checking if a stream is blocked [beam]

[via GitHub]

m-trieu commented on code in PR #34283:
URL: https://github.com/apache/beam/pull/34283#discussion_r2025841191


##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java:
##########
@@ -148,19 +171,101 @@ protected boolean hasPendingRequests() {
     @Override
     protected void startThrottleTimer() {}
 
-    public void testSend(Integer i)
-        throws ResettableThrowingStreamObserver.StreamClosedException,
-            WindmillStreamShutdownException {
+    public void testSend(Integer i) throws WindmillStreamShutdownException {
       trySend(i);
     }
 
     @Override
-    protected void sendHealthCheck() {}
+    protected void sendHealthCheck() {
+      numHealthChecks.incrementAndGet();
+    }
+
+    private void waitForHealthChecks(int expectedHealthChecks) {
+      int waitedMillis = 0;
+      while (numHealthChecks.get() < expectedHealthChecks) {
+        LOG.info(
+            "Waited for {}ms for {} health checks. Current health check count 
is {}.",
+            waitedMillis,
+            numHealthChecks.get(),
+            expectedHealthChecks);
+        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+      }
+    }
 
     @Override
     protected void appendSpecificHtml(PrintWriter writer) {}
 
     @Override
     protected void shutdownInternal() {}
   }
+
+  private static class TestCallStreamObserver extends 
CallStreamObserver<Integer> {
+    private static final Logger LOG = 
LoggerFactory.getLogger(AbstractWindmillStreamTest.class);
+    private final CountDownLatch sendBlocker = new CountDownLatch(1);
+    private final AtomicInteger numSends = new AtomicInteger();
+
+    private final boolean waitForSend;
+
+    private TestCallStreamObserver(boolean waitForSend) {

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscr...@beam.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org