remove processing/scheduling logic from StreamingDataflowWorker #31317

m-trieu · 2024-05-16T18:47:29Z

includes #31148

StreamingDataflowWorker will eventually just be constructing the harness runtime (different implementations for appliance, dispatched, fanout etc) and the main method.

R: @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 · 2024-05-16T19:34:33Z

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

R: @chamikaramj 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

An initial pass, didn't look much at tests yet

...-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/ReaderCache.java

...worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationState.java

...a/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ExecutionState.java

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

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

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

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

scwhittle · 2024-05-24T09:49:34Z

...a/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ExecutionState.java


 @AutoValue
-public abstract class ExecutionState {
+@Internal
+public abstract class ExecutionState implements AutoCloseable {


Can you add a comment here?
We have a bunch of context-y things for processing floating around. Would be good to note that this is per-computation and that it can be reused across bundles/work with resetting of things it contains.

Separately I wonder if we should merge this and StreamingModeExecutionStateContext?

I think it will be worth exploring cleaning up StreamingModeExecutionContext + ExecutionState

looks like streaming.worker.ExecutionState (there is another execution state StreamingModeExecutionState which extends DataflowExecutionState which extends core.metrics.ExecutionState which seems more for metric sampling)

There could be a lot of clean up here

Added a comment and TODO to clean up as well

ExecutionState is also specific to a computation so will add comment and rename accordingly.

...a/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ExecutionState.java

...g/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingCommitFinalizer.java

...org/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingWorkScheduler.java

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

...java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/ExecuteWorkResult.java

...src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationWorkExecutor.java

….start(...))

m-trieu · 2024-05-29T07:10:54Z

test failure is unrelated @scwhittle

scwhittle

Sorry for the delay, nothing too major left

scwhittle · 2024-05-29T07:41:11Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

    return backlogBytes;
  }

  public boolean workIsFailed() {
-    return workIsFailed.get();
+    return Optional.ofNullable(work).map(Work::isFailed).orElse(false);


work is not marked Nullable, can the ofNullable be removed?

is it ever possible that this is called when work is not set?

It seems like work should be marked Nullable since it's not set to non-null in constructor

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

scwhittle · 2024-05-29T07:50:07Z

...src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationWorkExecutor.java

+    closeWorkExecutor();
+  }
+
+  public final void closeWorkExecutor() {


comment? when should this be called? is it possible to reuse after this method?

Maybe we can drop this and just use ComputationWorkExexutor.close

Seems like it's just used in tests

removed, letting the tests just use ComputationWorkExexutor.invalidate()

scwhittle · 2024-05-29T07:56:40Z

...taflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java

+    public abstract String computationId();
+
+    /**
+     * {@link WindmillStream.GetDataStream} that connects to the backend Windmill worker handling


update comment to just say how to handle GetData requests (not necessarily a stream for appliance)

scwhittle · 2024-05-29T07:59:26Z

...taflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java

+     */
+    public abstract Consumer<Commit> workCommitter();
+
+    public abstract Optional<WindmillStream.GetDataStream> getDataStream();


add comment

is this just exposed so we can do heartbeat batching since we have keyedDataFetcher otherwise? If so perhaps it would be better to change to some more restricted interface in a follow up PR

yea since for refreshing work we need to group all active work by getDataStream

scwhittle · 2024-05-29T08:50:42Z

...orker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java

+    WorkCommitter workCommitter = mock(WorkCommitter.class);
+    doNothing().when(workCommitter).commit(any(Commit.class));
+    WindmillStream.GetDataStream getDataStream = mock(WindmillStream.GetDataStream.class);
+    when(getDataStream.requestKeyedData(anyString(), any()))


scwhittle · 2024-05-29T08:52:30Z

...g/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingCommitFinalizer.java

+  private static final Logger LOG = LoggerFactory.getLogger(StreamingCommitFinalizer.class);
+  private static final Duration DEFAULT_CACHE_ENTRY_EXPIRY = Duration.ofMinutes(5L);
+  private final Cache<Long, Runnable> onCommitFinalizedCache;
+  private final BoundedQueueExecutor workExecutor;


nit: rename throughout to just executor or finalizationExecutor

since we're not dealing with work here it's a little confusing to call workExecutor

...g/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingCommitFinalizer.java

scwhittle · 2024-05-29T09:00:09Z

...g/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingCommitFinalizer.java

+  }
+
+  /**
+   * Calls callbacks for WorkItem to mark that commit has been persisted (finalized) to the backing


I think the comment could be clearer how about something like

When this method is called, the commits associated with the provided finalizeIds have been successfully persisted in the backing state store. If the commitCallback for the finalizationId is still cached it is invoked.

scwhittle · 2024-05-29T09:00:51Z

...g/apache/beam/runners/dataflow/worker/windmill/work/processing/StreamingCommitFinalizer.java

+  }
+
+  /**
+   * Stores a map of user worker generated id's and callbacks to execute once a commit has been


how about being more specific "user worker generated finalization ids"
since we have lots of ids floating around :)

good idea done

scwhittle · 2024-05-29T15:50:43Z

I think the examples failure is unrelated so don't bother investigating it.

codecov · 2024-05-30T00:28:53Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 68.51%. Comparing base (5c30b1d) to head (8da5f40).
Report is 42 commits behind head on master.

❗ Current head 8da5f40 differs from pull request most recent head 570c4ef

Please upload reports for the commit 570c4ef to get more accurate results.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #31317      +/-   ##
============================================
- Coverage     68.55%   68.51%   -0.05%     
+ Complexity    14921    14823      -98     
============================================
  Files          2636     2636              
  Lines        222080   221689     -391     
  Branches      11825    11784      -41     
============================================
- Hits         152238   151881     -357     
+ Misses        63647    63629      -18     
+ Partials       6195     6179      -16

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

scwhittle

Just a couple more comments

...taflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/Work.java

scwhittle · 2024-05-30T08:06:32Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

    return backlogBytes;
  }

  public boolean workIsFailed() {
-    return workIsFailed.get();
+    return Optional.ofNullable(work).map(Work::isFailed).orElse(false);


It seems like work should be marked Nullable since it's not set to non-null in constructor

scwhittle · 2024-05-30T08:06:59Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

+ * {@link DataflowExecutionContext} for use in streaming mode. Contains cached readers and Beam
+ * state pertaining to a processing its owning computation. Can be reused across processing
+ * different WorkItems for the same computation.
+ */
 @SuppressWarnings({
  "nullness" // TODO(https://github.com/apache/beam/issues/20497)


maybe this can be removed to catch the missing work Nullable and perhaps other things?

Compiler will complain with under initialized error will update to prevent both

there are quite a bit of nullability issues not just with work nullability

added a comment to deal with this a follow up PR

I'm wondering if the separate state via biulder and then accessor instead of just local variables will lead to performance regression since this is on the hot path.

Since we don't actually clear the workScopedState between work items it seems that the null check will only help before start is ever called and then the additional machinery will be a no-op but more expensive.

Can you revert back to existing local variables for now? Perhaps we can improve nullability/correctness verification while merging this and the ComputationWorkExecutor and do a perf comparison. I don't want to make this big change riskier.

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

m-trieu · 2024-05-31T08:10:36Z

run java precommit

scwhittle · 2024-05-31T10:11:38Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

+ * {@link DataflowExecutionContext} for use in streaming mode. Contains cached readers and Beam
+ * state pertaining to a processing its owning computation. Can be reused across processing
+ * different WorkItems for the same computation.
+ */
 @SuppressWarnings({
  "nullness" // TODO(https://github.com/apache/beam/issues/20497)


I'm wondering if the separate state via biulder and then accessor instead of just local variables will lead to performance regression since this is on the hot path.

Since we don't actually clear the workScopedState between work items it seems that the null check will only help before start is ever called and then the additional machinery will be a no-op but more expensive.

Can you revert back to existing local variables for now? Perhaps we can improve nullability/correctness verification while merging this and the ComputationWorkExecutor and do a perf comparison. I don't want to make this big change riskier.

scwhittle

LGTM just going to run some internal tests before merging

scwhittle · 2024-05-31T19:58:51Z

Run Java PreCommit

github-actions bot added runners dataflow labels May 16, 2024

github-actions bot added the Next Action: Reviewers label May 16, 2024

scwhittle requested changes May 22, 2024

View reviewed changes

m-trieu force-pushed the mt-scheduler branch from 0d526e4 to 3af57e0 Compare May 23, 2024 07:26

scwhittle reviewed May 23, 2024

View reviewed changes

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java Outdated Show resolved Hide resolved

scwhittle requested changes May 24, 2024

View reviewed changes

m-trieu commented May 25, 2024

View reviewed changes

...src/main/java/org/apache/beam/runners/dataflow/worker/streaming/ComputationWorkExecutor.java Show resolved Hide resolved

m-trieu added 6 commits May 28, 2024 11:00

use work processing context in work processing

2a1288e

address cl comments

5debdc0

address cl comments

e5c91c7

break out work scheduling/processing from StreamingDataflowWorker

e8e7631

address PR comments

9b37bcb

address PR comments (remove overload of StreamingModeExecutionContext…

f43a44e

….start(...))

m-trieu force-pushed the mt-scheduler branch from 17e9609 to ed5d9ea Compare May 28, 2024 18:04

address PR comments

721bad3

m-trieu force-pushed the mt-scheduler branch from ed5d9ea to 721bad3 Compare May 28, 2024 18:29

scwhittle requested changes May 29, 2024

View reviewed changes

m-trieu force-pushed the mt-scheduler branch from 8da5f40 to efd55f4 Compare May 30, 2024 00:28

address PR comments

1248252

m-trieu force-pushed the mt-scheduler branch from efd55f4 to 1248252 Compare May 30, 2024 05:20

scwhittle requested changes May 30, 2024

View reviewed changes

address PR comments

5e91fca

m-trieu force-pushed the mt-scheduler branch from 35da731 to 5e91fca Compare May 31, 2024 05:43

scwhittle requested changes May 31, 2024

View reviewed changes

address PR comments revert state class in StreamingModeExecContext

570c4ef

scwhittle approved these changes May 31, 2024

View reviewed changes

scwhittle merged commit f93a67a into apache:master May 31, 2024
16 of 17 checks passed

remove processing/scheduling logic from StreamingDataflowWorker #31317

remove processing/scheduling logic from StreamingDataflowWorker #31317

Conversation

m-trieu commented May 16, 2024

GitHub Actions Tests Status (on master branch)

github-actions bot commented May 16, 2024

scwhittle left a 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 May 29, 2024

scwhittle left a 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scwhittle commented May 29, 2024

codecov bot commented May 30, 2024 • edited

Codecov Report

scwhittle left a 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 May 31, 2024

Choose a reason for hiding this comment

scwhittle left a comment

Choose a reason for hiding this comment

scwhittle commented May 31, 2024

codecov bot commented May 30, 2024 •

edited