add graceful restart mechanism for GetWorkStream to prevent DEADLINE_… #34367

m-trieu · 2025-03-20T21:24:40Z

GetWorkStream was previously terminating due to DEADLINE_EXCEEDED status leading to stream breaks and unnecessary retry spikes in the streaming backend.

Add GetWorkStream management loop that gracefully terminates and restarts the stream
Extend deadlines of direct streams to 1hr for less churn in restarting the streams internally
Add worker token to stream manager thread name

R: @scwhittle @acrites

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-20T22:06:17Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

…EXCEEDED; add long deadlines of 1hr to direct streams

m-trieu · 2025-03-23T21:32:30Z

R: @scwhittle

github-actions · 2025-03-23T21:33:47Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

m-trieu · 2025-03-24T15:11:56Z

assign set of reviewers

github-actions · 2025-03-24T15:13:19Z

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).

m-trieu · 2025-03-25T02:55:05Z

Run Java precommit

m-trieu · 2025-03-26T23:22:51Z

test failure is unrelated

scwhittle · 2025-03-27T10:17:40Z

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

+            streamingEngineStreamFactory.createDirectGetWorkStream(
+                connection,
+                withRequestBudget(getWorkRequest, getWorkBudget.get()),
+                streamingEngineThrottleTimers.getWorkThrottleTimer(),


should we just use the same throttle timer, heartbeat sender, and getdataclientfactory for each stream?

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

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

+                FixedStreamHeartbeatSender.create(getDataStream),
+                getDataClientFactory.apply(getDataStream),
+                workCommitter,
+                workItemScheduler);
    // 3 threads, 1 for each stream type (GetWork, GetData, CommitWork).
    this.streamStarter =
        Executors.newFixedThreadPool(


how about newCachedThreadPool? it seems like 2 of these threads are just for start() and then won't be used and we might as well have them go away.

scwhittle · 2025-03-27T10:24:05Z

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

-    if (started.get()) {
-      getWorkStream.setBudget(budget);
+    synchronized (activeGetWorkStream) {
+      GetWorkBudget budget = GetWorkBudget.builder().setItems(items).setBytes(bytes).build();


this build and setting on the atomic, could be outside synchronized block

scwhittle · 2025-03-27T10:25:41Z

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

+    synchronized (activeGetWorkStream) {
+      GetWorkBudget budget = GetWorkBudget.builder().setItems(items).setBytes(bytes).build();
+      getWorkBudget.set(budget);
+      if (isRunning.get()) {


remove running check and just use the null below? seems like if activeGetWorkStream is set it is ok to call and it's one less interleaving to think about

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

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

+        }
+
+      } catch (InterruptedException e) {
+        // continue until !isRunning.


Could we instead force that isRunning is set to false here? we don't expect interruptions to happen for any other reason.

scwhittle · 2025-03-27T10:31:51Z

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

+      try {
+        // Try to gracefully terminate the stream.
+        if (!newStream.awaitTermination(GET_WORK_STREAM_TTL_MINUTES, TimeUnit.MINUTES)) {
+          newStream.halfClose();


I think as we half-close here we probably want to create a new stream to take over.
That way we aren't idle while we're waiting for the termination.

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

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

+        }
+
+        // If graceful termination is unsuccessful, forcefully shutdown.
+        if (!newStream.awaitTermination(30, TimeUnit.SECONDS)) {


maybe better to increase this? If we lose getwork responses then windmill worker has to retry, if we can get them to flush with a little more time that seems fine.

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

.../org/apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcWindmillStreamFactory.java

@@ -189,6 +189,12 @@ private static <T extends AbstractStub<T>> T withDefaultDeadline(T stub) {
    return stub.withDeadlineAfter(DEFAULT_STREAM_RPC_DEADLINE_SECONDS, TimeUnit.SECONDS);
  }

+  private static <T extends AbstractStub<T>> T withLongDeadline(T stub) {


withDirectPathDeadline?

scwhittle · 2025-03-27T10:35:08Z

...ain/java/org/apache/beam/runners/dataflow/worker/streaming/harness/WindmillStreamSender.java

-  private final GetWorkStream getWorkStream;
+  private static final Logger LOG = LoggerFactory.getLogger(WindmillStreamSender.class);
+  private static final String STREAM_MANAGER_THREAD_NAME_FORMAT = "WindmillStreamManagerThread";
+  private static final int GET_WORK_STREAM_TTL_MINUTES = 45;


could note that this needs to be less than the deadline in the other file

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

...apache/beam/runners/dataflow/worker/windmill/client/grpc/observers/DirectStreamObserver.java

        ? "Waited "
            + totalSecondsWaited
            + "s which exceeds given deadline of "
-            + deadlineSeconds
+            + inactivityTimeout
            + "s for the outboundObserver to become ready meaning "
            + "that the stream deadline was not respected."


this seems like the wrong message if it isn't the stream deadline

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

m-trieu force-pushed the mt-getworkstream branch from d4fbaf0 to 362dd32 Compare March 20, 2025 21:25

add graceful restart mechanism for GetWorkStream to prevent DEADLINE_…

e746fd0

…EXCEEDED; add long deadlines of 1hr to direct streams

m-trieu force-pushed the mt-getworkstream branch from 362dd32 to e746fd0 Compare March 20, 2025 22:11

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

scwhittle requested changes Mar 27, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add graceful restart mechanism for GetWorkStream to prevent DEADLINE_… #34367

add graceful restart mechanism for GetWorkStream to prevent DEADLINE_… #34367

m-trieu commented Mar 20, 2025

github-actions bot commented Mar 20, 2025

m-trieu commented Mar 23, 2025

github-actions bot commented Mar 23, 2025

m-trieu commented Mar 24, 2025

github-actions bot commented Mar 24, 2025

m-trieu commented Mar 25, 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

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

scwhittle Mar 27, 2025

add graceful restart mechanism for GetWorkStream to prevent DEADLINE_… #34367

Are you sure you want to change the base?

add graceful restart mechanism for GetWorkStream to prevent DEADLINE_… #34367

Conversation

m-trieu commented Mar 20, 2025

GitHub Actions Tests Status (on master branch)

github-actions bot commented Mar 20, 2025

m-trieu commented Mar 23, 2025

github-actions bot commented Mar 23, 2025

m-trieu commented Mar 24, 2025

github-actions bot commented Mar 24, 2025

m-trieu commented Mar 25, 2025

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

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