do not block other threads from health checking if a stream is blocked #34283

m-trieu · 2025-03-13T19:22:16Z

Currently a single blocked stream can prevent health checks from multiple streams. Use the stream's internal executor to hand do the health check RPC in a way that does not block the health check scheduling thread.

R: @acrites @scwhittle

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2025-03-13T20:07:16Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @johnjcasey added as fallback since no labels match configuration

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

scwhittle · 2025-03-14T12:31:37Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java

-        logger.debug("Received exception sending health check.", e);
-      }
-    }
+  public final void maybeSendHealthCheck(Instant lastSendThreshold) {


I think that we might want to have the check if a health-check should be scheduled before scheduling on the executor.

Otherwise there may be more churn on the executor than necessary maybe adding to contention if maybeSendHealthCheck is called more often. We probably want to keep track if there is an active health-check with a member boolean as well to avoid piling up in the executor since maybeSendHealthCheck is no longer blocking and may be called more frequently than actually sending the health checks takes.

…ng, add tests.

m-trieu · 2025-03-19T19:05:29Z

ready for a another look @scwhittle

scwhittle · 2025-03-25T09:46:29Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java

+   *     synchronized block, we skip the attempt to send scheduled the health check.
+   */
+  public final void maybeSendHealthCheck(Instant lastSendThreshold) {
+    if (debugMetrics.getLastSendTimeMs() < lastSendThreshold.getMillis() && !isHealthCheckActive) {


maybe it should be !isHealthCheckScheduled and then set it to true here before putting on the executor?

if the executor gets backlogged we could still end up queuing a lot of things on the executor since currently we don't mark active until it schedules

done, opted to use an AtomicBoolean

scwhittle · 2025-03-25T09:50:11Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+    // Make sure the future completes.
+    sendFuture.get();
+
+    assertThat(testStream.numHealthChecks.get()).isEqualTo(1);


not sure if this is going to be flaky since lambda queued by trySend could possibly not have actually executed yet. Seems like it could be safer to loop while it is not yet 1.

scwhittle · 2025-03-25T09:50:24Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+    testStream.start();
+    ExecutorService sendExecutor = Executors.newSingleThreadExecutor();
+    Instant reportingThreshold = Instant.now().minus(Duration.millis(1));
+    Future<?> sendFuture =


I don't think you need to schedule this on a executor because maybeSendHealthCheck is no longer blocking.

scwhittle · 2025-03-25T09:51:44Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+              testStream.maybeSendHealthCheck(reportingThreshold);
+            });
+
+    callStreamObserver.unblockSend();


can we just start the test with it unblocked? Seems like you could just remove the sendFuture then and just do everything from the main test thread.

scwhittle · 2025-03-25T09:52:41Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+              // Set a really long reporting threshold.
+              Instant reportingThreshold = Instant.now().minus(Duration.standardHours(1));
+              // Should not send health checks since we just sent the above message.
+              testStream.maybeSendHealthCheck(reportingThreshold);


might need to sleep or verify the internal executor for the stream has nothing queud before you check no health-checks were sent

… and remove multithreading where possible

m-trieu · 2025-03-26T22:14:19Z

ready for a another look @scwhittle

scwhittle · 2025-03-27T09:58:45Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java

              }
+
+              // Ready to send another health check after we attempt the scheduled health check.
+              isHealthCheckScheduled.set(false);


how about showing if health check is scheduled in the appendSummaryHtml?

scwhittle · 2025-03-27T10:02:04Z

...ain/java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStream.java

              }
+
+              // Ready to send another health check after we attempt the scheduled health check.


seems like it would be a little safer to move the try to cover all of this lambda
so that we are sure to set it back to false.

try {
sycnrhonized () { ... maybe send }
} catch {
log
} finally {
isHealthCheckScheduled.set(false);
}

scwhittle · 2025-03-27T10:03:26Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+      throw new RuntimeException(e);
+    }
+
+    // Sleep a bit to give sendExecutor time to execute the send().


doesn't trySend send inline? can this be removed?

scwhittle · 2025-03-27T10:04:23Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+    callStreamObserver.waitForSend();
+    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);


// Sleep just to ensure a aysnc health check doesn't show up

scwhittle · 2025-03-27T10:07:32Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

+        int waitedMillis = 0;
+        while (!sendBlocker.await(100, TimeUnit.MILLISECONDS)) {
+          waitedMillis += 100;
+          LOG.info("Waiting from send for {}ms", waitedMillis);


I think sendBlocker is being used two different ways and is a little confusing.

How about naming this method to waitForSendUnblocked and logging here that you are waiting for send to be unblocked.
And then below in onNext instead of decrementing sendblocker, instead you increment a count of number of received and can add a new method to wait for N messages to be sent.

do not block other threads from health checking if a stream is blocked

6d76ad2

github-actions bot added runners dataflow labels Mar 13, 2025

github-actions bot added the Next Action: Reviewers label Mar 13, 2025

scwhittle requested changes Mar 14, 2025

View reviewed changes

m-trieu added 2 commits March 18, 2025 13:47

add boolean to check whether health check is active and skip scheduli…

81f3121

…ng, add tests.

fix sendHealthCheck() visibility

acb4515

scwhittle requested changes Mar 25, 2025

View reviewed changes

don't schedule a health check if there is one pending; simplify tests…

184223b

… and remove multithreading where possible

scwhittle requested changes Mar 27, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

do not block other threads from health checking if a stream is blocked #34283

do not block other threads from health checking if a stream is blocked #34283

m-trieu commented Mar 13, 2025

github-actions bot commented Mar 13, 2025

scwhittle Mar 14, 2025

m-trieu Mar 18, 2025

m-trieu commented Mar 19, 2025

scwhittle Mar 25, 2025

m-trieu Mar 25, 2025

scwhittle Mar 25, 2025

m-trieu Mar 26, 2025

scwhittle Mar 25, 2025

m-trieu Mar 26, 2025

scwhittle Mar 25, 2025

m-trieu Mar 26, 2025

scwhittle Mar 25, 2025

m-trieu Mar 26, 2025

m-trieu commented Mar 26, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

		}

		// Ready to send another health check after we attempt the scheduled health check.

do not block other threads from health checking if a stream is blocked #34283

Are you sure you want to change the base?

do not block other threads from health checking if a stream is blocked #34283

Conversation

m-trieu commented Mar 13, 2025

GitHub Actions Tests Status (on master branch)

github-actions bot commented Mar 13, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m-trieu commented Mar 19, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m-trieu commented Mar 26, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment