[Dataflow Streaming] Modify AbstractWindmillStream to separate logical stream from physical streams. by scwhittle · Pull Request #35327 · apache/beam

scwhittle · 2025-06-17T11:46:07Z

Currently this should be equivalent behavior but the separate physical stream handlers will allow for multiple distinct physical substreams when we want to cleanly terminate streams to avoid deadlines.

Add additional tests for the reconnection on stream errors (which already occurs). These could be committed separately if desired.

While at it change the deadline used for get data parsing timeout to be consistent with the deadline used for the response observers. Currently this won't change much but as we use a longer deadline for direct streams and update that in the future we will avoid these spurious errors.

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.

scwhittle · 2025-06-17T13:30:29Z

R: @arunpandianp @m-trieu

github-actions · 2025-06-17T13:31:30Z

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

arunpandianp · 2025-06-23T07:58:39Z

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

-    failureHandler.throwIfNonEmpty();
+    @Override
+    public void appendHtml(PrintWriter writer) {
+      writer.format("CommitWorkStream: %d pending", pending.size());


should we count only entries matching e.getValue().getKey() == this here?

arunpandianp · 2025-06-23T08:24:14Z

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

+        try {
+          BackOffUtils.next(Sleeper.DEFAULT, backoff);
+        } catch (IOException | InterruptedException ie) {
+          Thread.currentThread().interrupt();


why interrupt on IOException?

I'm also not sure why BackOffUtils.next has a throws IOException, it doesn't look like any implementation throws.

good point, separating here but will send you another PR to clean that up.

arunpandianp · 2025-06-23T08:35:28Z

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

  private final Deque<QueuedBatch> batches;

-  private final Map<Long, AppendableInputStream> pending;
+  private final Supplier<Integer> batchesDebugSizeSupplier;


thoughts on replacing this with the following?

`
private final ConcurrentLinkedDeque batches;

@SuppressWarnings("GuardedBy")
private int getBatchesSize() {
batches.size();
}`

I prefer doing it without requiring a suppression since it doesn't seem too complicated.

arunpandianp · 2025-06-23T08:42:05Z

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

  private static final long HEARTBEAT_REQUEST_ID = Long.MAX_VALUE;

-  private final ConcurrentMap<Long, PendingRequest> pending;
+  private final ConcurrentMap<Long, KV<CommitWorkPhysicalStreamHandler, PendingRequest>> pending =


It looks like we move all pending requests to the new steam on onNewStream, it is not clear to me if there'll be multiple CommitWorkPhysicalStreamHandlers be in use at the same time. Will it change in the following PRs?

Yes the goal of this PR is to set it up so that we can easily add the logic to half-close the physical stream before we hit the deadline exceed (without requiring to change all the uses of the logical streams to do it themselves).

To avoid causing disruptions to sending new requests we will start a new physical substream at that point. So in that case we don't want to resend the pending requests that are still active on the half-closing stream.

scwhittle · 2025-06-27T11:14:55Z

Having trouble reproducing the failures locally, still investigating.

m-trieu · 2025-06-29T00:47:19Z

Sorry for the delay! Will take a pass this week

arunpandianp · 2025-07-01T07:58:56Z

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

        } catch (RuntimeException e) {
-          // Catch possible exceptions to ensure that an exception for one commit does not prevent
+          // Catch possible exceptions to ensure that an exception for one commit does not
+          // prevent


need to fix format

arunpandianp · 2025-07-01T08:00:20Z

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

    synchronized (this) {
-      if (!prepareForSend(id, pendingRequest)) {
+      if (isShutdown) {
        pendingRequest.abort();


should we return here?

oops, didn't mean to remove that. I added more test coverage. It didn't matter before since the trySend would just fail and after shutdown things in pending didn't make a difference. But still good to fix.

arunpandianp · 2025-07-01T08:09:56Z

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

-    if (hasPendingRequests()) {
+  protected synchronized void sendHealthCheck() throws WindmillStreamShutdownException {
+    if (currentPhysicalStream != null && currentPhysicalStream.hasPendingRequests()) {
      trySend(HEALTH_CHECK_REQUEST);


not related to new code: will it be better to send health checks even when there are no pending requests?

These custom health-checks were more to make sure we werent' stuck waiting for responses on broken streams. Sending all the time could help detect errors on streams that are idle, reducing latency if the stream does need reconnecting and we wait until a request is sent. But for cases with load I don't think it woudl occur and always sending could increase the service load since I think we currently have a lot of idle streams (such as GetData that is never used). So I'd rather not change it unless we have more motivation.

…l stream from physical streams. Currently this should be equivalent behavior but the separate physical stream handlers will allow for multiple distinct physical substreams when we want to cleanly terminate streams to avoid deadlines. Add additional tests for the reconnection on stream errors (which already occurs). These could be committed separately if desired. While at it change the deadline used for get data parsing timeout to be consistent with the deadline used for the response observers. Currently this won't change much but as we use a longer deadline for direct streams and update that in the future we will avoid these spurious errors.

scwhittle · 2025-07-01T23:20:21Z

Run Java PreCommit

scwhittle · 2025-07-02T16:06:01Z

Run Java PreCommit

github-actions bot added runners dataflow labels Jun 17, 2025

scwhittle force-pushed the substream branch from 0711982 to 0f17fac Compare June 17, 2025 13:29

arunpandianp reviewed Jun 23, 2025

View reviewed changes

scwhittle force-pushed the substream branch from 36eff8c to f4aa0cc Compare June 27, 2025 11:40

scwhittle marked this pull request as ready for review June 30, 2025 18:15

arunpandianp reviewed Jul 1, 2025

View reviewed changes

scwhittle added 5 commits July 1, 2025 20:17

address comments

3c53e12

fix nullness, change retries

c34ba48

comments

67e3eac

address review comments

c53cd0d

scwhittle force-pushed the substream branch from f4aa0cc to c53cd0d Compare July 1, 2025 18:19

arunpandianp approved these changes Jul 1, 2025

View reviewed changes

fix spotless after rebase

c3f74b0

fix internal linter warnings

e86cf24

scwhittle merged commit 13e5921 into apache:master Jul 2, 2025
14 of 15 checks passed

scwhittle deleted the substream branch July 2, 2025 19:06

Conversation

scwhittle commented Jun 17, 2025

GitHub Actions Tests Status (on master branch)

Uh oh!

scwhittle commented Jun 17, 2025

Uh oh!

github-actions bot commented Jun 17, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

scwhittle commented Jun 27, 2025

Uh oh!

m-trieu commented Jun 29, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

scwhittle commented Jul 1, 2025

Uh oh!

scwhittle commented Jul 2, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants