Workflow streams user publish executor - #3025
Open
diwu-sf wants to merge 6 commits into
Open
Conversation
|
|
| toStop = ownedSchedulerLocked(); | ||
| } | ||
| if (toCancel != null) { | ||
| toCancel.cancel(false); |
Member
There was a problem hiding this comment.
After a background flush times out, this only cancels the periodic future when a user executor is supplied. Because the shared executor stays live, a later forceFlush=true or maxBatchSize trigger still submits backgroundFlush and sends the next buffered batch before throwDeferred surfaces the original error. I reproduced this with a user executor: let the first batch time out, clear the signal failure, then publish a second item with forceFlush=true; the second item is delivered. This also contradicts the new test comment that later items stay buffered until explicit flush or close. Please gate all triggered background flushes once deferredError is set, without shutting down the caller-owned executor, or otherwise surface the deferred error before another background send.
brianstrauch
requested changes
Aug 24, 2026
brianstrauch
left a comment
brianstrauch
left a comment
Member
There was a problem hiding this comment.
Blocking on the publish-executor timeout behavior described in the inline comment. A caller-owned executor remains able to run forced or size-triggered background flushes after a deferred FlushTimeoutException, allowing later data to be sent before the failure is surfaced.
StreamPublisher always created its own native platform thread executor to drive the background flush loop, one thread per publisher, with no way to share an executor across publishers. Applications that want to run flushes on virtual threads or a shared pool had no way to opt in. Add a constructor that accepts a ScheduledExecutorService. When one is supplied, the publisher never shuts it down: it only cancels the periodic flush task on close or on a deferred flush timeout, leaving the executor free for its other work. The default (no executor) behavior is unchanged: a lazily created single-thread executor owned and shut down by the publisher.
Every WorkflowStreamClient paid a dedicated platform thread for its publisher's flush loop, with no way to share an executor across clients even though WorkflowStreamClientOptions already allows one for the poll path. Applications that want virtual threads or one shared pool for many clients had no way to opt in. Add setPublishExecutor, mirroring setPollExecutor: the supplied executor drives the background flushes and the client never shuts it down — on close it only stops its own tasks. Default behavior is unchanged. Document the option in the module README and cover it with an integration test sharing one executor across two clients.
Mentioning virtual threads ties the docs to the newest JDKs even though any shared ScheduledExecutorService works. Describe the option in terms of sharing one executor across clients instead.
Cancelling the periodic task only stopped one of the two ways a background send starts: publish() still submitted backgroundFlush onto the executor on forceFlush or a full buffer. Previously the executor shutdown closed both paths, but a caller-owned executor must stay alive, so a triggered flush shipped the next batch before flush()/close() surfaced the deferred FlushTimeoutException. Track the stop in a loopStopped flag set with deferredError, checked in publish()'s trigger decision and at the top of backgroundFlush() so a task queued before the stop cannot send either. Gating on deferredError itself is not enough: throwDeferred() clears it, and the next triggered publish would resubmit — onto an already-shut-down executor in the client-owned case, which is where that path leaked RejectedExecutionException out of publish(). Both executor modes now behave the same. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Review follow-ups to the flush-timeout gating: - WorkflowStreamClient.close() ran publisher.close() outside try/finally, so the FlushTimeoutException its javadoc documents skipped subscription teardown and the owned poll executor shutdown, leaving pollers running forever. - backgroundFlush() checked loopStopped but not closed. A triggered flush already queued on a shared user executor could still signal after close() returned; shutdownNow() used to make that impossible. - ensureStartedLocked() left scheduleWithFixedDelay unguarded while the execute() below it was wrapped, so a user executor shut down by its owner threw RejectedExecutionException out of publish() after latching started. - The README implied one flush() call both drains and rethrows; flush() throws before sending anything, close() drains first. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The try/finally left three levels of teardown inline. Pull the driver loop and the poll executor shutdown into named methods so close() reads as its two steps; the executor block is moved, not changed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
diwu-sf
force-pushed
the
workflow-streams-user-publish-executor
branch
from
August 25, 2026 04:51
855dd2c to
32e3d8e
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What was changed
Every
StreamPublisherinstance was creating its own single thread executor to run the background flush. If the worker has high concurrency and many publishers, more and more platform threads will be created (and these mostly just idle).When using virtual threads, it's also not possible to override this executor with a shared one that uses virtual threads. Follow the pattern from workflow stream client and allow configuring a shared publish executor.
Why?
To reduce the number of idle platform threads created for each stream publisher instance.
Checklist
Tests added
Doc updated