Integrate direct path #31902

m-trieu · 2024-07-16T13:47:58Z

Add correct shutdown mechanics
remove deadline from direct path streaming RPCs
consume worker metadata more efficient (i.e remove polling)

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-07-16T14:34:54Z

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

R: @shunping 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).

Abacn · 2024-07-22T15:19:50Z

R: @scwhittle

github-actions · 2024-07-22T15:22:00Z

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

scwhittle

Not finished, mostly looking at stream stuff to help find what is causing stuckness

scwhittle · 2024-07-24T18:53:37Z

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

  /** Send a request to the server. */
  protected final void send(RequestT request) {
    lastSendTimeMs.set(Instant.now().getMillis());
    synchronized (this) {
+      // Check if we should send after we acquire the lock.
+      if (isShutdown()) {
+        LOG.warn("Send called on a shutdown stream.");


if this is possible, don't log as customers don't like warning logs and open issues about them

if this should not be possible, perhaps better to throw an exception so that we notice and fix it.

scwhittle · 2024-07-24T18:54:35Z

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

    }
  }

  /** Starts the underlying stream. */
  protected final void startStream() {
    // Add the stream to the registry after it has been fully constructed.
    streamRegistry.add(this);
-    while (true) {
+    while (!isShutdown.get()) {


I'd just remove this check since you do it first thing below

scwhittle · 2024-07-24T19:00:44Z

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

-          return;
-        }
+    private void tryRestartStream() {
+      if (!isShutdown()) {


this looks racy, we check shutdown above in isStreamDone with synchronization, but then if it is shutdown before here, we end up with an error but won't restart the stream or remove it from the registry.

I would remove this one

scwhittle · 2024-07-24T19:01:21Z

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

-        setLastError(error);
+    private synchronized boolean isStreamDone() {
+      if (isShutdown() || (clientClosed.get() && !hasPendingRequests())) {
+        streamRegistry.remove(AbstractWindmillStream.this);


I think this kind of side effect is confusing in method that just sounds like an accessor

how about maybeTeardownStream()?

Add a comment on return value

scwhittle · 2024-08-07T10:46:47Z

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

@@ -176,7 +180,7 @@ private StreamingDataflowWorker(
      DataflowWorkerHarnessOptions options,
      HotKeyLogger hotKeyLogger,
      Supplier<Instant> clock,
-      StreamingWorkerStatusReporter workerStatusReporter,
+      Function<Supplier<Long>, StreamingWorkerStatusReporter> streamingWorkerStatusReporterFactory,


should this be a functional interface so you can document? it's unclear what the long supplier is

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

scwhittle · 2024-08-07T11:47:14Z

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

+        () -> {
+          try {
+            send(extension);
+          } catch (IllegalStateException e) {


should we handle illegalstateexception internally? or rely on executeSafely to catch it?

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

scwhittle · 2024-08-07T11:56:39Z

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

@@ -182,6 +184,12 @@ protected void onResponse(StreamingCommitResponse response) {
    }
  }

+  @Override
+  protected void shutdownInternal() {
+    pending.values().forEach(pendingRequest -> pendingRequest.onDone.accept(CommitStatus.ABORTED));


I'm worried if pending has somethign inserted between iterating and clear

can you instead use an iterator where you remove as you go so everything removed is guaranteed to be aborted?

scwhittle · 2024-08-07T12:00:56Z

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

@@ -163,7 +165,7 @@ protected void onResponse(StreamingCommitResponse response) {
        continue;
      }
      PendingRequest done = pending.remove(requestId);
-      if (done == null) {
+      if (done == null && !isShutdown()) {


this is going to get nullptr exception below, instead move the isShutdown check to whether or not to log inside this if

scwhittle · 2024-09-12T10:30:27Z

...src/main/java/org/apache/beam/runners/dataflow/options/DataflowStreamingPipelineOptions.java

@@ -302,4 +301,12 @@ public Integer create(PipelineOptions options) {
      return streamingOptions.isEnableStreamingEngine() ? Integer.MAX_VALUE : 1;
    }
  }
+
+  /** EnableStreamingEngine defaults to false unless one of the two experiments is set. */


looks like just 1 experiment

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

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

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

...pache/beam/runners/dataflow/worker/windmill/client/grpc/WindmillStreamShutdownException.java

scwhittle · 2024-09-12T12:58:17Z

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

+   * Only send the next value if the phaser is not terminated by the time we acquire the lock since
+   * the phaser can be terminated at any time.
+   */
+  private void tryOnNext(T value) {


not sure these phaser checks are necessary since the outboundObserver itself should stop blocking for onNext if the notifier is terminated.

but if the phaser is terminated, we don't want to call outboundObserver.onNext() right?

Even if you do check here there is still race between checking the phaser and calling onNext regardless. Internally the outboundObserver is already observing the phaser termination via getPhase() (and also blocking respecting phaser termination) so the extra check is just mental overhead I think.

scwhittle · 2024-09-12T13:00:03Z

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

          return;
        }
      } catch (TimeoutException e) {
+        if (isReadyNotifier.isTerminated()) {


don't think you need this, awaitAdvanceInterruptibly will return -1 if it's terminated.

scwhittle · 2024-09-12T13:01:10Z

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

@@ -39,7 +39,9 @@
 @ThreadSafe
 public final class DirectStreamObserver<T> implements StreamObserver<T> {


maybe this change can be submitted separately? Would be nice to have a test for it showing the previous bug as well.

+1, I was thinking why DirectStreamObserver needs to be modified for direct path.

we need to remove the blocking outside of the lock or else terminating the stream will be held up by the deadline waiting for a response

scwhittle · 2024-09-12T13:05:01Z

...g/apache/beam/runners/dataflow/worker/windmill/work/budget/EvenGetWorkBudgetDistributor.java

@@ -67,17 +70,18 @@ public <T extends GetWorkBudgetSpender> void distributeBudget(
      GetWorkBudgetSpender getWorkBudgetSpender = streamAndDesiredBudget.getKey();
      GetWorkBudget desired = streamAndDesiredBudget.getValue();
      GetWorkBudget remaining = getWorkBudgetSpender.remainingBudget();
-      if (isBelowFiftyPercentOfTarget(remaining, desired)) {
+      if (isBelowFiftyPercentOfTarget(remaining, desired) && isActiveWorkBudgetAware) {


this is unclear to me. It seems like in one case the adjustment is addtiive and another is resetting

Wouldn't we want to only increase by the desired-remaining in either case?

whether its overriding or additive is depending on the internal implementation of GetWorkBudgetSpender.adjustBudget()

based on the current implementation that this is reflecting, we do not account for remaining budget here it is handled internally in adjustBudget.

scwhittle

didn't get through everything but sending comments I had

scwhittle · 2024-09-13T08:41:29Z

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

+            + "<th>Active For</th>"
+            + "<th>State</th>"
+            + "<th>State Active For</th>"
+            + "<th>Produced By</th>"


nit: produced is confusing since we use that for shuffle terminology. How about just "Backend"

scwhittle · 2024-09-13T08:44:58Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

-    getWorkBudgetRefresher.requestBudgetRefresh();
+    getWorkBudgetDistributor.distributeBudget(newStreams.values(), totalGetWorkBudget);
+
+    // Close the streams outside the lock.


yeah I don't know if we need to block on closing streams. I think we might want to block on creating the new ones.

one possible thing is that if we are leaking stuff and they never actually close we might not know. But if we have the stream registry I think we'd see that. Or we can add some logging to closeAllStreams that it is taking a long time.

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

scwhittle · 2024-09-13T09:11:42Z

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

+    private final Supplier<StreamObserver<RequestT>> requestObserverSupplier;
+
+    @GuardedBy("this")
+    private volatile @Nullable StreamObserver<RequestT> delegateRequestObserver;


You don't need both, synchronization also ensures that memory changes are viewed by other threads. Volatile just ensures one fields's changes are visible to other threads.

This seems like a good overview: https://blogs.oracle.com/javamagazine/post/java-thread-synchronization-volatile-final-atomic-deadlocks

scwhittle · 2024-09-13T09:12:22Z

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

+    }
+
+    @Override
+    public synchronized void onCompleted() {


rm synchronized?

still synchronized

scwhittle · 2024-09-13T11:59:37Z

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

-    }
+    streamRegistry.stream()
+        .sorted(Comparator.comparing(AbstractWindmillStream::backendWorkerToken))
+        .collect(


I missed the multimap bit. I agree the grouping is nice. However isn't the sorting then unnecessary if we're putting it in a map right away?

scwhittle · 2024-09-13T12:01:39Z

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

@@ -264,24 +284,18 @@ protected void startThrottleTimer() {

  @Override
  public void adjustBudget(long itemsDelta, long bytesDelta) {


can we name this setBudget?

scwhittle · 2024-09-13T12:06:43Z

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

+      } catch (AppendableInputStream.InvalidInputStreamStateException
+          | VerifyException
+          | CancellationException e) {
+        handleShutdown(request);


Can we make shutdown handling consistent?
we throw an exception for shutdown here but if we don't run the loop due to isShutdown we just return from this function without doing anything.

we do throw a WindmillStreamShutdownException at the bottom if we exit the loop

scwhittle · 2024-09-13T12:07:10Z

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

+  private void handleShutdown(QueuedRequest request) {
+    if (isShutdown()) {
+      throw new WindmillStreamShutdownException(
+          "Cannot send request=[" + request + "] on closed stream.");


if we keep this, should we pass in the exception above to add as a suppressed exception?

scwhittle · 2024-09-13T12:10:13Z

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

+      }
+    }
+
+    ImmutableList<String> createStreamCancelledErrorMessage() {


add parameter for the limit or just inline above? We're building up a possibly big list just to ignore most of it.

m-trieu · 2024-09-20T06:51:04Z

@scwhittle ready for another look! I will resolve the merge conflicts and rebase

m-trieu · 2024-09-27T07:25:36Z

@scwhittle FanOutStreamingEngineWorkerHarnessTest.testOnNewWorkerMetadata_redistributesBudget() was flaky. fixed!

scwhittle

Sorry I didn't finish up reviewing and will be on vacation next week. Perhaps you could have Arun take a look.

In particular I need some more time to go over the stream changes as that affects the dispatcher path too.

scwhittle · 2024-09-26T11:53:47Z

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

+                    ? WindmillStreamPool.create(
+                        1, GET_DATA_STREAM_TIMEOUT, windmillServer::getDataStream)
+                    : getDataStreamPool);
+        statusPagesBuilder


can the status page stuff be moved after this block so it can be shared for the SE paths? It seems like same stuff is duplicated for direct and not

the only similarity is the DebugCapture manager I can consolidate that

consolidated in builder construction

moved creation to after harness intialization block

scwhittle · 2024-09-26T12:10:56Z

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

@@ -402,14 +455,15 @@ public static StreamingDataflowWorker fromOptions(DataflowWorkerHarnessOptions o
        options,
        new HotKeyLogger(),
        clock,
-        workerStatusReporter,
+        workerStatusReporterFactory,


should we just pass in the builder prepopulated instead of a factory?

wanted to constrain the interface to not allow callers to modify the other members
but if passing in a builder is preferred i can do that

scwhittle · 2024-09-26T12:19:04Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

+  private void consumeWorkerMetadata(WindmillEndpoints windmillEndpoints) {
+    synchronized (metadataLock) {
+      // Only process versions greater than what we currently have to prevent double processing of
+      // metadata.


add comemnt: the consumer is single-threaded so we maintain ordering.

scwhittle · 2024-09-26T12:21:17Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

+      // Only process versions greater than what we currently have to prevent double processing of
+      // metadata.
+      if (windmillEndpoints.version() > metadataVersion) {
+        metadataVersion = windmillEndpoints.version();


maybe it woudl be good to have pendingMetadataVersion (updated here) and activeMetadataVersion (which you update after consuming). That could help debugging since it could show if we're stuck on some old version.

done
we could add that information to the status pages (probably separate PR).

we only check pendingMetadataVersion then since pending == active if there are no updates pending

scwhittle · 2024-09-26T12:24:51Z

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

@@ -155,9 +155,11 @@ void closeAllStreams() {

  @Override
  public void adjustBudget(long itemsDelta, long bytesDelta) {


name setBudget as well?
update variables to not be delta, adjustment if it is just a total budget request

scwhittle · 2024-09-26T12:26:56Z

...worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/WindmillEndpoints.java

@@ -123,6 +124,8 @@ private static Optional<HostAndPort> tryParseDirectEndpointIntoIpV6Address(
            directEndpointAddress.getHostAddress(), (int) endpointProto.getPort()));
  }

+  public abstract long version();


// The metadata version increases with every modification.

scwhittle · 2024-09-26T12:33:07Z

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

          // Keep trying to create the stream.
        }
      }
    }
+
+    // We were never able to start the stream, remove it from the stream registry.


// Otherwise it is removed when closed.

scwhittle · 2024-09-26T12:35:08Z

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

-  public void shutdown() {
+  public final void shutdown() {
+    // Don't lock here as isShutdown checks are used in the stream to free blocked
+    // threads or as exit conditions to loops.
    if (isShutdown.compareAndSet(false, true)) {
      requestObserver()


I don't think requestObserver is thread-safe though? do we need to synchronize it?

requestObserver() is thread safe with its own internal synchronization.

scwhittle · 2024-09-26T12:35:28Z

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

+    }
+
+    @Override
+    public synchronized void onCompleted() {


still synchronized

scwhittle · 2024-09-27T11:10:53Z

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

    synchronized (this) {
      pending.put(id, pendingRequest);
      for (int i = 0;
-          i < serializedCommit.size();
+          i < serializedCommit.size() && !isShutdown();


I'd remove the shutdown check since it's not consistent with other paths and probably won't matter if the send just won't do anything anyway.

arunpandianp · 2024-10-04T01:30:01Z

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

+    @Nullable ChannelzServlet channelzServlet = null;
+    Consumer<PrintWriter> getDataStatusProvider;
+    Supplier<Long> currentActiveCommitBytesProvider;
+    if (isDirectPathPipeline(options)) {


Does this have to be set during pipeline creation? Thoughts on supporting on/off based on job settings?

sync'd offline

eventually will be propogated in workerMetadata, but there are some immutable settings that the VM needs to have that supports this, so we check here or fail.

… Future proofs this logic for direct path.

… have it implement WorkProvider interface. Move class to windmill/work/provider directory, update visibility for dependent classes and move tests, add GetWorkBudgetOwnerInterface

…ct path

… WindmillEndpoints and don't process any version that is older than the current version in FanOutStreamingEngineWorkerHarness

…a_redistributesBudget()

…ady 600 seconds anyway. Move DEFAULT_STREAM_RPC_DEADLINE_SECONDS to where it is being used and remove references in tests

arunpandianp · 2024-10-09T07:23:55Z

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

+                // IsolationChannel will create and manage separate RPC channels to the same
+                // serviceAddress via calling the channelFactory, else just directly return the
+                // RPC channel.
+                workerOptions.getUseWindmillIsolatedChannels()


We could choose to not support this for direct path and add it later if needed.
If we are keeping it, i think we need to use a different option to avoid confusing with the usage and rollout of the these flags in cloud path. In cloud path isolated channels are enabled based on this flag or job setting.

Thinking more, will it be better to default to isolated channels enabled? That'll be the closest to what we'll have in cloud path after #32782

arunpandianp

Still need to look at the Streams classes, sending some more comments I've.

arunpandianp · 2024-10-10T07:36:18Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

    started = true;
  }

  public ImmutableSet<HostAndPort> currentWindmillEndpoints() {
-    return connections.get().windmillConnections().keySet().stream()
+    return connections.get().windmillStreams().keySet().stream()
        .map(Endpoint::directEndpoint)
        .filter(Optional::isPresent)
        .map(Optional::get)


Will filtering out IPV6 below prevent direct path endpoints from showing up in channelz?

we don't need that filter i am going to remove the IPV6 since we don't use that anymore

arunpandianp · 2024-10-10T08:15:27Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

+  }
+
+  private synchronized CompletableFuture<ImmutableMap<Endpoint, WindmillStreamSender>>
+      createAndStartNewStreams(Collection<Endpoint> newWindmillConnections) {


Suggested change

createAndStartNewStreams(Collection<Endpoint> newWindmillConnections) {

createAndStartNewStreams(ImmutableSet<Endpoint> newWindmillConnections) {

arunpandianp · 2024-10-10T08:16:32Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

+        MoreFutures.allAsList(
+            newWindmillConnections.stream()
+                .map(
+                    connection ->
+                        MoreFutures.supplyAsync(
+                            () ->
+                                Pair.of(
+                                    connection,
+                                    Optional.ofNullable(currentStreams.get(connection))
+                                        .orElseGet(
+                                            () -> createAndStartWindmillStreamSender(connection))),


Suggested change

MoreFutures.allAsList(

newWindmillConnections.stream()

.map(

connection ->

MoreFutures.supplyAsync(

() ->

Pair.of(

connection,

Optional.ofNullable(currentStreams.get(connection))

.orElseGet(

() -> createAndStartWindmillStreamSender(connection))),

MoreFutures.allAsList(

newWindmillEndpoints.stream()

.map(

endpoint ->

MoreFutures.supplyAsync(

() ->

Pair.of(

endpoint,

Optional.ofNullable(currentStreams.get(endpoint))

.orElseGet(

() -> createAndStartWindmillStreamSender(endpoint))),

arunpandianp · 2024-10-10T08:17:13Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

-      WindmillConnection connection) {
-    // Initially create each stream with no budget. The budget will be eventually assigned by the
-    // GetWorkBudgetDistributor.
+  private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint connection) {


Suggested change

private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint connection) {

private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint endpoint) {

arunpandianp · 2024-10-10T08:24:27Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

+        previousMetadataVersion,
+        activeMetadataVersion);


previousMetadataVersion and activeMetadataVersion are same here.

arunpandianp · 2024-10-10T08:38:33Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

+    closeStaleStreams(
+        newWindmillEndpoints.windmillEndpoints(), connections.get().windmillStreams());
+    ImmutableMap<Endpoint, WindmillStreamSender> newStreams =
+        createAndStartNewStreams(newWindmillEndpoints.windmillEndpoints()).join();


We can defer the join() to be after creating the globalDataStreams.

global data streams are created in createAndStartNewStreams

well not really created
we pass a factory to thr StreamGetDataClient

and it will get created whenever the user code fetches side input

arunpandianp · 2024-10-10T08:48:33Z

...pache/beam/runners/dataflow/worker/streaming/harness/FanOutStreamingEngineWorkerHarness.java

-      WindmillConnection connection) {
-    // Initially create each stream with no budget. The budget will be eventually assigned by the
-    // GetWorkBudgetDistributor.
+  private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint connection) {
    WindmillStreamSender windmillStreamSender =


Why do we need a dispatcher fallback in createWindmillStub?

for fallback
if we don't pass in direct endpoint we will just fallback to dispatcher stub

at some point we were thinking about supporting dispatcher as a passthrough proxy with the user worker telling it which backend to hit and this would also allow that

arunpandianp · 2024-10-10T08:55:52Z

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

@@ -156,29 +162,47 @@ public void sendHealthCheck() {
  protected void onResponse(StreamingCommitResponse response) {
    commitWorkThrottleTimer.stop();

-    RuntimeException finalException = null;
+    @Nullable RuntimeException failure = null;


Create a new Exception here and attach all failures as suppressed?

Want to avoid marking failure from one request as suppressed of another.

arunpandianp · 2024-10-10T08:58:28Z

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

+          if (failure == null) {
+            failure = e;
+          } else {
+            failure.addSuppressed(e);


There could be 1000s of requests in the queue, do we want to add all of them here? Maybe only the finalException or a subset is enough?

It is important that we track it some how? maybe keep account of the error/exception type
it's possible that some might have a real issue but others won't

actually maybe a subset is good

this is an error consuming the commit since at this point windmill is acking that the commit was either successful or failed. so maybe we can just record the status + failure or have a map<<status, exception>, count> and just log that

opted to record last 10 errors in detail and have a counter of <status, throwable.class> so we can keep track of any weird behavior

arunpandianp · 2024-10-10T09:01:13Z

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

@@ -187,13 +211,14 @@ protected void startThrottleTimer() {
    commitWorkThrottleTimer.start();
  }

-  private void flushInternal(Map<Long, PendingRequest> requests) {
+  private void flushInternal(Map<Long, PendingRequest> requests) throws InterruptedException {


do we need throws InterruptedException here?

we do since it is a checked exception thrown by one of the methods inside flushInternal
we handle it in flush()

wanted to make it clear the flushInternal is blocking()
if we get interrupted we interrupt the thread, but always clear the queue and queuedBytes (in the finally block)

…e StreamingEngineConnectionsState to StreamingEngineBackends

m-trieu changed the title ~~Mt dp streams~~ Add correct shutdown mechanics for direct path Jul 16, 2024

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

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

m-trieu force-pushed the mt-dp-streams branch 3 times, most recently from 01e8586 to ff287f2 Compare August 6, 2024 12:25

scwhittle requested changes Aug 7, 2024

View reviewed changes

m-trieu force-pushed the mt-dp-streams branch 2 times, most recently from bbf2778 to 1e02463 Compare September 12, 2024 05:06

scwhittle requested changes Sep 12, 2024

View reviewed changes

m-trieu force-pushed the mt-dp-streams branch 2 times, most recently from 4820b9f to d03b0bc Compare September 13, 2024 05:19

scwhittle requested changes Sep 13, 2024

View reviewed changes

m-trieu force-pushed the mt-dp-streams branch 3 times, most recently from d95d1b5 to 9bad2d7 Compare September 20, 2024 03:53

m-trieu force-pushed the mt-dp-streams branch from 9bad2d7 to 08a9c1e Compare September 24, 2024 07:52

m-trieu changed the title ~~Add correct shutdown mechanics for direct path~~ Integrate direct path Sep 26, 2024

m-trieu force-pushed the mt-dp-streams branch from 4464453 to d0a199d Compare September 26, 2024 20:48

scwhittle requested changes Sep 27, 2024

View reviewed changes

m-trieu force-pushed the mt-dp-streams branch 3 times, most recently from 19b50d9 to 745d87c Compare October 3, 2024 21:31

arunpandianp reviewed Oct 4, 2024

View reviewed changes

m-trieu added 24 commits October 9, 2024 18:32

add HeartbeatSender interface

8c1ca87

address PR comments

9f42751

move side input state API into Work instance w/ keyed state fetching.…

e623fc7

… Future proofs this logic for direct path.

don't let FanOutWorkRefreshClient exceptions crash the user worker

0735a0c

address PR comments

998c07a

address PR comments

659d5aa

add WorkProvider interface and have StreamingEngine and Appliance impls

554b73e

change StreamingEngineClient to FanOutStreamingEngineWorkProvider and…

a7a9efa

… have it implement WorkProvider interface. Move class to windmill/work/provider directory, update visibility for dependent classes and move tests, add GetWorkBudgetOwnerInterface

add WorkProvider interface and have StreamingEngine and Appliance impls

52c1fec

change StreamingEngineClient to FanOutStreamingEngineWorkProvider and…

83655fa

… have it implement WorkProvider interface. Move class to windmill/work/provider directory, update visibility for dependent classes and move tests, add GetWorkBudgetOwnerInterface

add WorkProvider interface and have StreamingEngine and Appliance impls

e27fde2

change StreamingEngineClient to FanOutStreamingEngineWorkProvider and…

9bf73c4

… have it implement WorkProvider interface. Move class to windmill/work/provider directory, update visibility for dependent classes and move tests, add GetWorkBudgetOwnerInterface

add stream management and correct stream shutdown mechanisms for dire…

9beeec9

…ct path

integrate direct path components

0d60a18

address PR comments

361ec3c

simplify budgeting logic for GrpcDirectGetWorkStream

fd6549e

fix stuckness on range movements

36015f8

address PR comments, fix broken tests

14025db

address PR comments

2144cd9

remove thread polling for new metadata; propogate metadata version to…

92414f7

… WindmillEndpoints and don't process any version that is older than the current version in FanOutStreamingEngineWorkerHarness

deflake FanOutStreamingEngineWorkerHarnessTest.testOnNewWorkerMetadat…

b09ba8f

…a_redistributesBudget()

address PR comments

3db3b7f

address PR comments

06836da

remove MAX_WAIT_SECONDS and just use the deadline passed in, its alre…

6df1adf

…ady 600 seconds anyway. Move DEFAULT_STREAM_RPC_DEADLINE_SECONDS to where it is being used and remove references in tests

m-trieu force-pushed the mt-dp-streams branch from a59733f to 6df1adf Compare October 10, 2024 01:40

arunpandianp reviewed Oct 10, 2024

View reviewed changes

address PR comments, remove unused IPv6 WindmillServiceAddress, renam…

4f5b381

…e StreamingEngineConnectionsState to StreamingEngineBackends

m-trieu force-pushed the mt-dp-streams branch from de3b016 to 4f5b381 Compare October 11, 2024 17:02

m-trieu mentioned this pull request Oct 12, 2024

Enable external callers to start a WindmillStream #32763

Open

3 tasks

		@@ -39,7 +39,9 @@
		@ThreadSafe
		public final class DirectStreamObserver<T> implements StreamObserver<T> {

		@@ -264,24 +284,18 @@ protected void startThrottleTimer() {

		@Override
		public void adjustBudget(long itemsDelta, long bytesDelta) {

		@@ -155,9 +155,11 @@ void closeAllStreams() {

		@Override
		public void adjustBudget(long itemsDelta, long bytesDelta) {

	createAndStartNewStreams(Collection<Endpoint> newWindmillConnections) {
	createAndStartNewStreams(ImmutableSet<Endpoint> newWindmillConnections) {

	private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint connection) {
	private WindmillStreamSender createAndStartWindmillStreamSender(Endpoint endpoint) {

Integrate direct path #31902

Are you sure you want to change the base?

Integrate direct path #31902

Conversation

m-trieu commented Jul 16, 2024 • edited Loading

R: @scwhittle

GitHub Actions Tests Status (on master branch)

github-actions bot commented Jul 16, 2024

Abacn commented Jul 22, 2024

github-actions bot commented Jul 22, 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

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

m-trieu commented Sep 20, 2024

m-trieu commented Sep 27, 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

m-trieu commented Jul 16, 2024 •

edited

Loading