add shutdown and start mechanics to windmill streams #32774

m-trieu · 2024-10-14T21:20:50Z

Start and closing Windmill streams are currently via halfClose() and on stream creation. Implementations were previously created and returned in a "started" state usually after the stream has already sent the initial headers to open the connection to the backend servers.

Starting in the current state prevents us from being able to start the stream "lazily". And closing allows other blocking stream operations to prevent streams from being able to be closed (stalling at times up to 10-20 minutes).

Add start() flexibility to the WindmillStream API by allowing external callers to start the stream themselves.
Add shutdown() capability to allow the stream to receive a shutdown signal, that is idempotent and does not block (or is blocked by) other blocking stream operations.

This is especially important in direct path mode where the user worker manages the fan out to the backend.

in terms of implementation, similar to WindmillStream.shutdown(), WindmillStream.start()'s behavior will only execute once during the lifetime of the WindmillStream object. Subsequent calls to start() and shutdown() will do nothing.

R: @arunpandianp @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-10-14T22:07:01Z

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

m-trieu · 2024-10-15T00:23:33Z

assign set of reviewers

github-actions · 2024-10-15T00:24:47Z

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

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

arunpandianp · 2024-10-15T07:30:27Z

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

@@ -216,12 +207,18 @@ static FanOutStreamingEngineWorkerHarness forTesting(
    return fanOutStreamingEngineWorkProvider;
  }

-  @SuppressWarnings("ReturnValueIgnored")
+  @SuppressWarnings("FutureReturnValueIgnored")


assign the future to a variable named unusedFuture and remove the suppression?

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

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

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

...er/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/WindmillStream.java

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

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

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

arunpandianp · 2024-10-15T16:02:36Z

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

        throw new RuntimeException(e);
      } finally {
        pending.remove(request.id());
      }
    }
+
+    // If we have exited the loop here, the stream has been shutdown. Cancel the response stream.
+    request.getResponseStream().cancel();


do we need this? only the parseFn.parse above was reading the responseStream and nothing should be blocked on the response stream at this point?

not needed, just wanted to cancel the stream incase there was anything dangling

If it is not needed, lets remove it. To catch anything dangling, we can add a check and throw

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

m-trieu · 2024-10-29T09:49:00Z

back to you thanks @scwhittle

scwhittle · 2024-10-30T13:47:20Z

...n/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableStreamObserver.java

+  private @Nullable StreamObserver<T> delegateStreamObserver;
+
+  /**
+   * Indicates that the request observer should no longer be used. Attempts to perform operations on


move comment next to poison method?

...n/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableStreamObserver.java

scwhittle · 2024-10-30T13:50:28Z

...rc/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/StreamDebugMetrics.java

+/** Records stream metrics for debugging. */
+@ThreadSafe
+final class StreamDebugMetrics {
+  private final AtomicInteger restartCount = new AtomicInteger();


can these just be synchronized ints as well instead of atomic? WE're not doing anything expensive under lock so it doesn't seem like it needs separate consideration.

scwhittle · 2024-10-30T13:52:52Z

...rc/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/StreamDebugMetrics.java

+    lastRestartTime = DateTime.now();
+  }
+
+  synchronized long startTimeMs() {


getStartTimeMs?

ditto for lastSendTimeMs

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

scwhittle · 2024-10-30T14:18:11Z

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

+  /**
+   * Used to guard {@link #start()} and {@link #shutdown()} behavior.
+   *
+   * @implNote Do not hold when performing IO. If also locking on {@code this} in the same context,


if you acquire this before (this) lock, then this mutex could be blocked by I/O because we perform IO beneath this lock.

If this is supposed to be lightweight it seems like it should be acquired after (this) to avoid that.

scwhittle · 2024-10-30T14:18:39Z

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

+  @GuardedBy("this")
+  private boolean streamClosed;
+
+  private volatile boolean isShutdown;


can this be guarded by shutdown lock? the accessor method can synchronize

scwhittle · 2024-10-30T14:18:55Z

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

+  private boolean streamClosed;
+
+  private volatile boolean isShutdown;
+  private volatile boolean started;


can this be guarded by one of hte locks instead of volatile?

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

m-trieu · 2024-11-01T01:00:20Z

back to you @scwhittle thank you!

scwhittle · 2024-11-05T10:19:16Z

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

        throw new IllegalStateException("Send called on a client closed stream.");
      }

-      requestObserver().onNext(request);
+      try {
+        verify(!Thread.holdsLock(shutdownLock), "shutdownLock should not be held during send.");


I would remove this since it seems likely expensive.
Instead you could verify with a test:

setup requestObserver that blocks until notified

one thread calls send and starts blocking

main test thread calls shutdown() and verifies the method returns

main test thread unblocks the requestObserver

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

...n/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableStreamObserver.java

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

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

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

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

scwhittle · 2024-11-06T10:14:55Z

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

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

-    RuntimeException finalException = null;
+    CommitCompletionException failures = new CommitCompletionException();


nit: how about a builder for this since we won't want to add exceptions in other places.

then the final line of method can be

builder.throwIfNonEmpty();
and interrnally it builds exception and throws if needed?

I was thinking the builder would not be an exception itself, and then the exception would just be a simple class without mutating methods.

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

scwhittle · 2024-11-06T10:22:51Z

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

-      queuedBytes = 0;
-      queue.clear();
+      try {
+        if (!hasReceivedShutdownSignal()) {


can we remove? it is handled in stream methods already and doing it here leaves a gap between shutdown check and flush anyway.

scwhittle · 2024-11-06T10:25:23Z

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

    send(StreamingGetDataRequest.newBuilder().setHeader(jobHeader).build());
-    if (clientClosed.get()) {
+    if (clientClosed && !hasReceivedShutdownSignal()) {


remove? above send will fail if shutdown
though I think we can also guarantee at higher level that onNewStream isn't called if shutdown

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

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

scwhittle · 2024-11-08T08:52:49Z

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

-        return true;
-      }
+    private boolean maybeTeardownStream() {
+      synchronized (AbstractWindmillStream.this) {


just make a synchronized method?

fixed by moving it out of the inner class
GuardedBy check was failing since reference to this was ambiguous

scwhittle · 2024-11-08T08:53:58Z

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

+    private boolean maybeTeardownStream() {
+      synchronized (AbstractWindmillStream.this) {
+        if (isShutdown || (clientClosed && !hasPendingRequests())) {
+          streamRegistry.remove(AbstractWindmillStream.this);


why doesn't just "this" work?

fixed by moving it out of the inner class
GuardedBy check was failing since reference to this was ambiguous

scwhittle · 2024-11-08T08:57:15Z

...rc/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/StreamDebugMetrics.java

@@ -59,6 +59,9 @@ final class StreamDebugMetrics {
  @GuardedBy("this")
  private DateTime shutdownTime = null;


mark nullable and above

I think you still need to fix the over exemption of windmill classes from checker.
#30183

scwhittle · 2024-11-08T09:08:15Z

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

        // If the stream was stopped due to a resource exhausted error then we are throttled.
-        if (status != null && status.getCode() == Status.Code.RESOURCE_EXHAUSTED) {
+        if (status.getCode() == Status.Code.RESOURCE_EXHAUSTED) {


nit. I would maybe keep this in the onError case, as the other stuff is more just logs/debug page and this is more functional.

scwhittle · 2024-11-08T10:14:07Z

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

-    while (true) {
+  private <ResponseT> ResponseT issueRequest(QueuedRequest request, ParseFn<ResponseT> parseFn)
+      throws WindmillStreamShutdownException {
+    while (!isShutdownLocked()) {


rm? below needs to handle it anyway

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

scwhittle · 2024-11-08T10:16:13Z

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

  }

-  private void queueRequestAndWait(QueuedRequest request) throws InterruptedException {
+  private synchronized void handleShutdown(QueuedRequest request, Throwable cause)


nit: name throwIfShutdown?

scwhittle · 2024-11-08T10:21:34Z

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

-      request.addToStreamingGetDataRequest(builder);
-    }
-    return builder.build();
+  private synchronized void verify(boolean condition, String message) {


can this be removed? it seems you have shutdown check first in cases I see (and if not I think it would be clearer to have it as part of the check where it is than hidden in method that doesn't sound like it examines shutdown)

scwhittle · 2024-11-08T10:21:51Z

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

+    Verify.verify(condition || isShutdown, message);
+  }
+
+  private synchronized boolean isShutdownLocked() {


rm if you remove above

scwhittle

some of previous comments were not resolved, please look through them too. Thanks!

scwhittle · 2024-11-12T13:35:50Z

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

    requestObserver.poison();
+    isShutdown = true;


we should remove this (and the suppress)

shouldn't the poison prevent the blocking beneath the mutex? and then the below lock will be acquired soon?

Setting it to true outside the mutex will break invariants that are easier to think about if it is strictly guarded by. (and it breaks logic below we'll never run shutdownInternal)

if we do need it for something it seems like we could have a separate volatile shutdownRequested boolean. But I'd prefer to figure out what gets stuck with the current code and fix it instead because it is confusing to have two.

cleaneed up was supposed to stay within the sync block

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

scwhittle · 2024-11-12T13:43:23Z

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

@@ -472,7 +473,8 @@ private void flushResponse() {
                    responseObserver.onNext(responseBuilder.build());
                  } catch (Exception e) {
                    // Stream is already closed.
-                    System.out.println("trieu: " + e);
+                    LOG.warn("trieu: ", e);


rm debug logs

removed this waas for debugging

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

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

scwhittle · 2024-11-12T14:19:27Z

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

    synchronized (lock) {
+      isClosed = true;


ie if (!isClosed) check to all of these

m-trieu · 2024-11-13T10:25:36Z

still need to add some more tests to DirectStreamObserverTest
addressed the other comments

thanks!
@scwhittle

scwhittle · 2024-11-13T12:16:47Z

...rg/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java

@@ -121,7 +121,7 @@ public void onNext(T t) throws StreamClosedException, WindmillStreamShutdownExce

      try {
        delegate.onError(e);
-      } catch (RuntimeException ignored) {
+      } catch (IllegalStateException ignored) {


add the ignored to the exception too? just in case we catch something unexpected and it helps debugging?

scwhittle · 2024-11-13T12:19:01Z

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

@@ -188,7 +188,9 @@ protected void onResponse(StreamingGetDataResponse chunk) {

    for (int i = 0; i < chunk.getRequestIdCount(); ++i) {
      AppendableInputStream responseStream = pending.get(chunk.getRequestId(i));
-      verify(responseStream != null, "No pending response stream");
+      synchronized (this) {


avoid synchronizing if not necessary and you need to handle the null case or you will just get an exception in following line.

if (responseStream == null) {
sychronized (this) { verify(isShutdown); }
continue;
}

scwhittle · 2024-11-13T12:21:44Z

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

      outboundObserver.onCompleted();
    }
  }

+  private void markClosedOrThrow() {


just make method synchronized (though with suggestion below, probably easier to just duplicate in onCompleted/onError and use the same if block.

scwhittle · 2024-11-13T12:25:29Z

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

      outboundObserver.onCompleted();
    }
  }

+  private void markClosedOrThrow() {
+    synchronized (lock) {
+      Preconditions.checkState(!isClosed);


this is going to throw if we have sequence

T1: onNext()
T2: terminate()
T1: onError()

It seems like that could happen if we're terminating from other threads than the one driving the observer generally. We could have a separate bool tracking if userClosed or not, and change this exception to be based upon that as that is using the class wrong. having a terminate before/during a onCompleted/onError doesn't necessarily seem like misuse and I think we should avoid throwing an exception.

scwhittle · 2024-11-13T12:26:47Z

...java/org/apache/beam/runners/dataflow/worker/windmill/client/AbstractWindmillStreamTest.java

@@ -104,6 +106,10 @@ public void setMessageCompression(boolean b) {}
            () ->
                assertThrows(WindmillStreamShutdownException.class, () -> testStream.testSend(1)));
    testStream.shutdown();
+
+    // Sleep a bit to give sendExecutor time to execute the send().
+    Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);


sleep for less, 100ms maybe? Tests take long enough already

scwhittle · 2024-11-13T12:29:31Z

.../apache/beam/runners/dataflow/worker/windmill/client/grpc/GrpcGetDataStreamRequestsTest.java

@@ -142,7 +144,8 @@ public void testQueuedBatch_notifyFailed_throwsWindmillStreamShutdownExceptionOn
                assertThrows(
                    WindmillStreamShutdownException.class,
                    queuedBatch::waitForSendOrFailNotification));
-
+    // Wait a few seconds for the above future to get scheduled and run.
+    Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);


ditto how about 100ms

scwhittle · 2024-11-13T12:29:53Z

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

+            });
+
+    // Sleep a bit to allow future to run.
+    Uninterruptibles.sleepUninterruptibly(5, TimeUnit.SECONDS);


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

m-trieu · 2024-11-15T07:57:17Z

back to you @scwhittle thanks!

scwhittle · 2024-11-15T09:01:45Z

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

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

+  public final boolean isEmpty() {
+    return equals(none());


How about changing none() to return some singleton instead of building every time if we might be calling empty a lot.

removed but made none a singleton

scwhittle · 2024-11-15T09:03:43Z

...rg/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java

@@ -104,6 +104,10 @@ synchronized void poison() {
    }
  }

+  synchronized boolean hasReceivedPoisonPill() {


nit: how about isPoisoned?

Poison pill was a special element we added to queues etc, it's a little confusing to use the term here to me.

scwhittle · 2024-11-15T09:06:07Z

...rg/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java

@@ -125,6 +129,8 @@ public void onNext(T t) throws StreamClosedException, WindmillStreamShutdownExce
        // If the delegate above was already terminated via onError or onComplete from another
        // thread.
        logger.warn("StreamObserver was previously cancelled.", e);
+      } catch (RuntimeException ignored) {
+        logger.warn("StreamObserver was unexpectedly cancelled.", e);


I was thinking above you should log both e and the currently ignored exception

ditto here. Also this log message is a little confusing how about
"encountered error {} when cancelling due to error"

scwhittle · 2024-11-15T09:10:56Z

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

@@ -89,6 +90,9 @@ public void onNext(T value) throws StreamObserverCancelledException {
            throw new StreamObserverCancelledException("StreamObserver was terminated.");
          }

+          // We close under "lock", so this should never happen.


needs note about terminating phaser first, update below too.

// We close under "lock" after terminating, so if we didn't observer termination above
// we shouldn't be closed.

scwhittle · 2024-11-15T09:15:49Z

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

@@ -162,24 +170,23 @@ public void onNext(T value) throws StreamObserverCancelledException {
  public void onError(Throwable t) {
    isReadyNotifier.forceTermination();
    synchronized (lock) {
-      markClosedOrThrow();
-      outboundObserver.onError(t);
+      if (!isClosed) {


I think we should have
isUserClosed = onError or onCompleted called on this object
isClosed = onError or onCompleted called on outboundObserver
Could name isClosed and isOutboundClosed or something if clearer. Woudl be good to comment above too.

I think that onError/onCompleted should be like:

check(!isUserClosed);
isUserClosed = true;
if (!isClosed) {
outboundObserver.onError(t);
isClosed = true;
}

enforcing that only one of them is called. And terminate should be like

if (!isClosed) {
outboundObserver.onError(t);
isClosed = true;
}

done
also caught the illegal state exception in AbstractWindmillStream#halfClose

scwhittle · 2024-11-15T09:25:08Z

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

+                        () -> closeStreamSender(sender.endpoint(), sender), windmillStreamManager))
+            .collect(Collectors.toList());
+
+    return ImmutableList.<CompletableFuture<Void>>builder()


Do we want to return the list of futures or woudl it better to return a single future by using CompletableFuture.allOf(..)

scwhittle · 2024-11-15T09:34:56Z

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

  }

  @Override
  public String backendWorkerToken() {
    return backendWorkerToken;
  }

+  @SuppressWarnings("GuardedBy")


remove suppression

scwhittle · 2024-11-15T09:39:58Z

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

+
+  /** Returns true if the stream was torn down and should not be restarted internally. */
+  private synchronized boolean maybeTearDownStream() {
+    if (requestObserver.hasReceivedPoisonPill()


what is this receivedPoisonPill check guarding against?
It is racy because we poison outside of the synchronized block so we could have

T1: notices unrelated stream failure, passes this check and isn't poisoned, starts calling onNewStream
T2: calls shutdown, poisons request observer
T1: calls requestObserver.reset() gets exception due to poison.

Instead of checking the poison here, it seems like we should just handle the exception due to reset failing as that covers both cases.

done, realized we weren't calling break on the WindmillStreamShutdownException in startStream()

scwhittle · 2024-11-15T09:43:22Z

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

+    }
+    commitWorkStream.shutdown();
+
+    Set<Windmill.CommitStatus> commitStatuses = new HashSet<>();


ping on unresolved comment here

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

m-trieu force-pushed the mt-start-stream branch 2 times, most recently from 9591a9a to ce57880 Compare October 14, 2024 23:35

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

arunpandianp reviewed Oct 15, 2024

View reviewed changes

m-trieu force-pushed the mt-start-stream branch 3 times, most recently from f05b41e to b66c086 Compare October 15, 2024 17:42

github-actions bot added python java go build model io kafka and removed python java go build model io kafka labels Oct 15, 2024

m-trieu force-pushed the mt-start-stream branch from b66c086 to 881399e Compare October 15, 2024 17:49

m-trieu mentioned this pull request Oct 15, 2024

Simplify budget distribution logic and new worker metadata consumption #32775

Merged

3 tasks

m-trieu force-pushed the mt-start-stream branch from 881399e to 8618174 Compare October 15, 2024 18:25

m-trieu force-pushed the mt-start-stream branch from 213e804 to 1b819bc Compare October 29, 2024 09:46

address PR comments

77969d0

m-trieu force-pushed the mt-start-stream branch from 1b819bc to 77969d0 Compare October 29, 2024 09:50

don't block on rendering status pages

5a68e0f

m-trieu force-pushed the mt-start-stream branch from d49d905 to 5a68e0f Compare October 29, 2024 18:42

scwhittle requested changes Oct 30, 2024

View reviewed changes

address PR comments

3fb76ac

m-trieu force-pushed the mt-start-stream branch from 685d386 to 3fb76ac Compare November 1, 2024 00:59

scwhittle requested changes Nov 5, 2024

View reviewed changes

address PR comments

3fafefa

scwhittle requested changes Nov 6, 2024

View reviewed changes

address PR comments

ab34387

scwhittle requested changes Nov 8, 2024

View reviewed changes

address PR comments part 1

4a9a863

scwhittle requested changes Nov 12, 2024

View reviewed changes

address PR comments

95a19f4

scwhittle requested changes Nov 13, 2024

View reviewed changes

m-trieu force-pushed the mt-start-stream branch 2 times, most recently from 64e61d9 to 42836f4 Compare November 13, 2024 22:00

address PR comments

f75df0f

m-trieu force-pushed the mt-start-stream branch from 42836f4 to f75df0f Compare November 14, 2024 09:00

scwhittle requested changes Nov 15, 2024

View reviewed changes

address PR comments

74e503d

scwhittle approved these changes Nov 19, 2024

View reviewed changes

scwhittle merged commit 5229c93 into apache:master Nov 19, 2024
16 of 17 checks passed

		@@ -59,6 +59,9 @@ final class StreamDebugMetrics {
		@GuardedBy("this")
		private DateTime shutdownTime = null;

add shutdown and start mechanics to windmill streams #32774

add shutdown and start mechanics to windmill streams #32774

Conversation

m-trieu commented Oct 14, 2024

R: @arunpandianp @scwhittle

GitHub Actions Tests Status (on master branch)

github-actions bot commented Oct 14, 2024

m-trieu commented Oct 15, 2024

github-actions bot commented Oct 15, 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

m-trieu commented Oct 29, 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 1, 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

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

m-trieu commented Nov 13, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

m-trieu commented Nov 15, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment