Integrate direct path with StreamingDataflowWorker code path #32778

m-trieu · 2024-10-14T23:58:35Z

Add direct path code branch

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-10-15T00:50:05Z

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

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

github-actions · 2024-10-30T12:14:14Z

Reminder, please take a look at this pr: @johnjcasey

github-actions · 2024-11-04T12:14:24Z

Assigning new set of reviewers because Pr has gone too long without review. 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)

github-actions · 2024-11-12T12:14:21Z

Reminder, please take a look at this pr: @johnjcasey

scwhittle · 2024-10-15T12:10:37Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+        : streamingStatusPages;
+  }
+
+  private static ChannelzServlet createChannelZServlet(


nit: don't capitalize the Z

scwhittle · 2024-10-15T12:17:07Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -118,6 +130,8 @@ public final class StreamingDataflowWorker {
   */
  public static final int MAX_SINK_BYTES = 10_000_000;

+  public static final String STREAMING_ENGINE_USE_JOB_SETTINGS_FOR_HEARTBEAT_POOL =


move back where it was and make private?

scwhittle · 2024-10-15T12:23:45Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -441,6 +520,8 @@ public static StreamingDataflowWorker fromOptions(DataflowWorkerHarnessOptions o
    WindmillServerStub windmillServer;
    ComputationStateCache computationStateCache;
    GrpcWindmillStreamFactory windmillStreamFactory;
+    ConfigFetcherComputationStateCacheAndWindmillClient.Builder builder =


seems inconsistent to set dispatcher on builder but have the separate variables for the rest.

How about removing the varialbes and setting on the builder? It seems you can remove some nesting and return the build() result early in some cases like SE

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

scwhittle · 2024-11-19T11:49:56Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+      }
+      LOG.warn(
+          "DirectPath is currently only supported with IPv6 networking stack. Defaulting to"
+              + " CloudPath.");


How about mentioning the service option to use or link to docs if some

scwhittle · 2024-11-19T11:52:00Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+        StreamingDataflowWorker.class.getSimpleName());
+  }
+
+  private static ChannelCachingStubFactory createStubFactory(


name createFanoutStubFactory since just used there and we don't want unconditional isolation (yet) in other case?

scwhittle · 2024-11-19T11:59:41Z

...apache/beam/runners/dataflow/worker/streaming/harness/StreamingWorkerStatusReporterTest.java

-            (threadName) -> Executors.newSingleThreadScheduledExecutor(),
-            DEFAULT_HARNESS_REPORTING_PERIOD,
-            DEFAULT_PER_WORKER_METRICS_PERIOD);
+        StreamingWorkerStatusReporter.builder()


can you make a testReporterBuilder() in this test that sets all these defaults?

m-trieu · 2024-11-19T22:00:42Z

back to you @scwhittle thanks!

github-actions · 2024-11-22T12:14:48Z

Assigning new set of reviewers because Pr has gone too long without review. 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)

scwhittle · 2024-11-25T16:35:46Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

+          .setWindmillStreamFactory(windmillStreamFactory)
+          .setWindmillServer(
+              GrpcWindmillServer.create(options, windmillStreamFactory, dispatcherClient))
+          .build();
    } else {


can remove else to reduce nesting since above always returns

scwhittle · 2024-11-25T16:39:18Z

...ava/org/apache/beam/runners/dataflow/worker/streaming/harness/SingleSourceWorkerHarness.java

@@ -74,7 +75,8 @@ public final class SingleSourceWorkerHarness implements StreamingWorkerHarness {
      StreamingWorkScheduler streamingWorkScheduler,
      Runnable waitForResources,
      Function<String, Optional<ComputationState>> computationStateFetcher,
-      GetWorkSender getWorkSender) {
+      GetWorkSender getWorkSender,
+      Supplier<Long> throttleTimeSupplier) {


can we make a functional interface instead of a supplier? It's not clear that the supplier should get+reset, generally Supplier seems like it would just get.

Seems like this could then take supplier to vend but also could implement the same interface

scwhittle · 2024-11-25T16:40:54Z

...n/java/org/apache/beam/runners/dataflow/worker/streaming/harness/StreamingWorkerHarness.java

@@ -25,4 +25,6 @@ public interface StreamingWorkerHarness {
  void start();

  void shutdown();
+
+  long getAndResetThrottleTime();


ditto could make StreamingWorkerHarness implement the ThrottlingTimerInterface (or whatever you name it).

scwhittle · 2024-11-25T18:29:00Z

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

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker.streaming.harness;


put in same throttling package as ThrottleTimer.java

scwhittle · 2024-11-25T18:29:30Z

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

+ */
+@Internal
+@FunctionalInterface
+public interface ThrottledTimeTracker {


this has a 'd' in Throttled but all the instances are just throttleTimeTracker. Make consistent

github-actions bot added runners dataflow labels Oct 14, 2024

github-actions bot added the Next Action: Reviewers label Oct 15, 2024

m-trieu commented Oct 18, 2024

View reviewed changes

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java Show resolved Hide resolved

m-trieu force-pushed the mt-fix-dp-experiment branch from 45fdb16 to da13655 Compare October 22, 2024 22:53

github-actions bot added the slow-review label Oct 30, 2024

github-actions bot removed the slow-review label Nov 4, 2024

github-actions bot added the slow-review label Nov 12, 2024

m-trieu added 3 commits November 19, 2024 00:20

Integrate direct path with StreamingDataflowWorker code path

50d94e0

address PR comments

b7dba9e

rebase and fix tests

395abf1

m-trieu force-pushed the mt-fix-dp-experiment branch from 036e3a1 to 395abf1 Compare November 19, 2024 08:47

scwhittle requested changes Nov 19, 2024

View reviewed changes

address PR comments

1c895f5

github-actions bot removed the slow-review label Nov 22, 2024

scwhittle requested changes Nov 25, 2024

View reviewed changes

address PR comments

a93bae5

m-trieu force-pushed the mt-fix-dp-experiment branch from d17e6a6 to a93bae5 Compare November 25, 2024 18:14

scwhittle requested changes Nov 25, 2024

View reviewed changes

address PR comments

978b660

scwhittle approved these changes Nov 26, 2024

View reviewed changes

scwhittle merged commit aa21e4a into apache:master Nov 26, 2024
17 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Integrate direct path with StreamingDataflowWorker code path #32778

Integrate direct path with StreamingDataflowWorker code path #32778

m-trieu commented Oct 14, 2024

github-actions bot commented Oct 15, 2024

github-actions bot commented Oct 30, 2024

github-actions bot commented Nov 4, 2024

github-actions bot commented Nov 12, 2024

scwhittle Oct 15, 2024

m-trieu Nov 19, 2024

scwhittle Oct 15, 2024

m-trieu Nov 19, 2024

scwhittle Oct 15, 2024

m-trieu Nov 19, 2024

scwhittle Nov 19, 2024

m-trieu Nov 19, 2024

scwhittle Nov 19, 2024

m-trieu Nov 19, 2024

scwhittle Nov 19, 2024

m-trieu Nov 19, 2024

m-trieu commented Nov 19, 2024

github-actions bot commented Nov 22, 2024

scwhittle Nov 25, 2024

m-trieu Nov 25, 2024

scwhittle Nov 25, 2024

m-trieu Nov 25, 2024

scwhittle Nov 25, 2024

m-trieu Nov 25, 2024

scwhittle Nov 25, 2024

m-trieu Nov 26, 2024

scwhittle Nov 25, 2024

m-trieu Nov 26, 2024

Integrate direct path with StreamingDataflowWorker code path #32778

Integrate direct path with StreamingDataflowWorker code path #32778

Conversation

m-trieu commented Oct 14, 2024

GitHub Actions Tests Status (on master branch)

github-actions bot commented Oct 15, 2024

github-actions bot commented Oct 30, 2024

github-actions bot commented Nov 4, 2024

github-actions bot commented Nov 12, 2024

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

m-trieu commented Nov 19, 2024

github-actions bot commented Nov 22, 2024

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