From e7c6d454b252651ce6e83f5f6d9e8ffcc6b43c17 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Thu, 20 Aug 2026 15:29:55 -0700 Subject: [PATCH 1/3] Let StreamPublisher run its flush loop on a user-supplied executor 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. --- .../internal/StreamPublisher.java | 73 ++++++++++-- .../workflowstreams/StreamPublisherTest.java | 106 ++++++++++++++++++ 2 files changed, 167 insertions(+), 12 deletions(-) diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java index f07e7095ad..3498ad4105 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java @@ -11,7 +11,9 @@ import java.util.UUID; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; /** * Owns the client-side publish path: it buffers published values, batches them, and sends each @@ -36,6 +38,8 @@ public interface SignalFunction { private final long batchIntervalMs; private final int maxBatchSize; private final long maxRetryDurationMs; + // When null, the publisher creates a single-thread executor it owns and shuts down in close(). + @Nullable private final ScheduledExecutorService userExecutor; private final Object stateLock = new Object(); private List buffer = new ArrayList<>(); @@ -46,7 +50,12 @@ public interface SignalFunction { private boolean started; private boolean closed; private FlushTimeoutException deferredError; + // The executor driving the flush loop once started; the owned one when no user executor was + // supplied. Guarded by stateLock. private ScheduledExecutorService scheduler; + // The periodic flush tick, tracked so it can be cancelled without shutting down a user-supplied + // executor. Guarded by stateLock. + private ScheduledFuture flushTask; /** Serializes doFlush so concurrent callers send sequentially. */ private final Object flushLock = new Object(); @@ -57,12 +66,31 @@ public StreamPublisher( Duration batchInterval, int maxBatchSize, Duration maxRetryDuration) { + this(signal, dataConverter, batchInterval, maxBatchSize, maxRetryDuration, null); + } + + /** + * @param executor drives the background flush loop (the periodic ticks and the flushes triggered + * by a full buffer or {@code forceFlush}). When non-null the caller owns its lifecycle and it + * is never shut down by this publisher, so many publishers can share one executor (e.g. a + * virtual-thread executor); when null a single-thread executor is created lazily, owned by + * this publisher, and shut down by {@link #close}. Flushes block while signaling the + * workflow, so each in-flight flush occupies an executor thread for the duration of the send. + */ + public StreamPublisher( + SignalFunction signal, + DataConverter dataConverter, + Duration batchInterval, + int maxBatchSize, + Duration maxRetryDuration, + @Nullable ScheduledExecutorService executor) { this.signal = signal; this.dataConverter = dataConverter; this.publisherId = UUID.randomUUID().toString().replace("-", "").substring(0, 16); this.batchIntervalMs = batchInterval.toMillis(); this.maxBatchSize = maxBatchSize; this.maxRetryDurationMs = maxRetryDuration.toMillis(); + this.userExecutor = executor; } /** @@ -97,15 +125,20 @@ private void ensureStartedLocked() { return; } started = true; - scheduler = - Executors.newSingleThreadScheduledExecutor( - r -> { - Thread t = new Thread(r, "temporal-workflow-stream-publisher"); - t.setDaemon(true); - return t; - }); - scheduler.scheduleWithFixedDelay( - this::backgroundFlush, batchIntervalMs, batchIntervalMs, TimeUnit.MILLISECONDS); + if (userExecutor != null) { + scheduler = userExecutor; + } else { + scheduler = + Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r, "temporal-workflow-stream-publisher"); + t.setDaemon(true); + return t; + }); + } + flushTask = + scheduler.scheduleWithFixedDelay( + this::backgroundFlush, batchIntervalMs, batchIntervalMs, TimeUnit.MILLISECONDS); } private void backgroundFlush() { @@ -114,10 +147,15 @@ private void backgroundFlush() { } catch (FlushTimeoutException e) { // The pending batch was dropped and can't be recovered. Stash the error so // flush/close surface it and stop the loop. + ScheduledFuture toCancel; ScheduledExecutorService toStop; synchronized (stateLock) { deferredError = e; - toStop = scheduler; + toCancel = flushTask; + toStop = ownedSchedulerLocked(); + } + if (toCancel != null) { + toCancel.cancel(false); } if (toStop != null) { toStop.shutdown(); @@ -226,18 +264,24 @@ public void flush() { /** * Stops the background flush loop and drains any remaining items, surfacing a deferred {@link - * FlushTimeoutException} from a prior background failure. + * FlushTimeoutException} from a prior background failure. A user-supplied executor is never shut + * down; only the periodic flush task is cancelled, leaving the executor free for its other work. */ public void close() { + ScheduledFuture toCancel; ScheduledExecutorService toStop; synchronized (stateLock) { if (closed) { return; } closed = true; - toStop = scheduler; + toCancel = flushTask; + toStop = ownedSchedulerLocked(); } + if (toCancel != null) { + toCancel.cancel(false); + } if (toStop != null) { toStop.shutdownNow(); try { @@ -259,6 +303,11 @@ public void close() { throwDeferred(); } + /** Returns the executor to shut down on stop, or null when a user executor must be left alone. */ + private ScheduledExecutorService ownedSchedulerLocked() { + return userExecutor == null ? scheduler : null; + } + private void throwDeferred() { synchronized (stateLock) { if (deferredError != null) { diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/StreamPublisherTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/StreamPublisherTest.java index 7c8cabb6e0..04eeb58ee2 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/StreamPublisherTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/StreamPublisherTest.java @@ -11,6 +11,10 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Assert; import org.junit.Test; @@ -20,19 +24,31 @@ public class StreamPublisherTest { /** Records sent batches; when {@code failure} is set, sending throws it instead. */ private static class RecordingSignal implements StreamPublisher.SignalFunction { final List signals = new ArrayList<>(); + final List threads = new ArrayList<>(); volatile RuntimeException failure; + int attempts; @Override public synchronized void send(PublishInput input) { + attempts++; if (failure != null) { throw failure; } + threads.add(Thread.currentThread().getName()); signals.add(input); } synchronized List recorded() { return new ArrayList<>(signals); } + + synchronized List threads() { + return new ArrayList<>(threads); + } + + synchronized int attempts() { + return attempts; + } } private static StreamPublisher newPublisher( @@ -65,6 +81,24 @@ private static void eventually(Duration timeout, Runnable assertion) throws Inte } } + private static ScheduledExecutorService newNamedExecutor(String name) { + return Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r, name); + t.setDaemon(true); + return t; + }); + } + + /** Proves the executor is alive and still accepting work (i.e. was not shut down). */ + private static void assertExecutorStillRunsTasks(ScheduledExecutorService executor) + throws InterruptedException { + Assert.assertFalse(executor.isShutdown()); + CountDownLatch ran = new CountDownLatch(1); + executor.execute(ran::countDown); + Assert.assertTrue(ran.await(5, TimeUnit.SECONDS)); + } + @Test public void testFlushSendsBufferedItems() { RecordingSignal signal = new RecordingSignal(); @@ -238,4 +272,76 @@ public void testFlushTimeoutAfterMaxRetryDuration() throws InterruptedException publisher.close(); } + + /** + * A user-supplied executor drives both the flushes triggered by {@code forceFlush} and the + * periodic ticks, and close() neither shuts it down nor interrupts its other work: the caller + * owns its lifecycle, so many publishers can share one executor. + */ + @Test + public void testUserExecutorDrivesFlushesAndSurvivesClose() throws InterruptedException { + RecordingSignal signal = new RecordingSignal(); + ScheduledExecutorService user = newNamedExecutor("user-publish-executor"); + StreamPublisher publisher = + new StreamPublisher( + signal, + DC, + Duration.ofMillis(50), + 0, + WorkflowStreamConstants.DEFAULT_MAX_RETRY_DURATION, + user); + + publisher.publish("t", "a", true); // immediate trigger runs on the user executor + eventually(Duration.ofSeconds(5), () -> Assert.assertEquals(1, signal.recorded().size())); + Assert.assertEquals("user-publish-executor", signal.threads().get(0)); + + publisher.publish("t", "b", false); // only a periodic tick can send it + eventually(Duration.ofSeconds(5), () -> Assert.assertEquals(2, signal.recorded().size())); + Assert.assertEquals("user-publish-executor", signal.threads().get(1)); + + // close() drains synchronously on the caller thread and must leave the executor alone. + publisher.publish("t", "c", false); + publisher.close(); + Assert.assertEquals(3, signal.recorded().size()); + assertExecutorStillRunsTasks(user); + user.shutdownNow(); + } + + /** + * After a background flush exceeds the max retry duration, the periodic task on a user executor + * is cancelled (not the executor) — so once the failure clears, later items stay buffered until + * an explicit flush or close drains them, and close() surfaces the deferred timeout. + */ + @Test + public void testUserExecutorTaskCancelledAfterFlushTimeout() throws InterruptedException { + RecordingSignal signal = new RecordingSignal(); + signal.failure = new RuntimeException("boom"); + ScheduledExecutorService user = newNamedExecutor("user-publish-executor"); + StreamPublisher publisher = + new StreamPublisher(signal, DC, Duration.ofMillis(20), 0, Duration.ofMillis(1), user); + + publisher.publish("t", "a", false); + // The first tick attempts the send and fails transiently, leaving the batch pending; + // the next tick exceeds the 1ms retry window and defers a FlushTimeoutException. + eventually(Duration.ofSeconds(5), () -> Assert.assertEquals(1, signal.attempts())); + Thread.sleep(300); + + signal.failure = null; + publisher.publish("t", "b", false); + // The cancelled task must not deliver "b": well past the interval, nothing is sent. + Thread.sleep(300); + Assert.assertTrue(signal.recorded().isEmpty()); + + try { + publisher.close(); + Assert.fail("unreachable"); + } catch (FlushTimeoutException expected) { + } + // close() still drains the buffered item on the caller thread, and the user executor + // was cancelled, not shut down. + Assert.assertEquals(1, signal.recorded().size()); + Assert.assertEquals("b", decodeItem(signal.recorded().get(0), 0)); + assertExecutorStillRunsTasks(user); + user.shutdownNow(); + } } From c4fb1f6e132406a9c18d7704fa0f7345c072e8a2 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Thu, 20 Aug 2026 15:32:33 -0700 Subject: [PATCH 2/3] Expose publishExecutor on WorkflowStreamClientOptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- contrib/temporal-workflowstreams/README.md | 6 ++ .../workflowstreams/WorkflowStreamClient.java | 6 +- .../WorkflowStreamClientOptions.java | 34 ++++++++++- .../workflowstreams/SubscribeTest.java | 57 +++++++++++++++++++ 4 files changed, 99 insertions(+), 4 deletions(-) diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index ad630c89b7..b71404d5b0 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -112,6 +112,11 @@ Items are buffered and flushed automatically every batch interval (default 2s), when the buffer reaches the max batch size, on `forceFlush`, on an explicit `flush()`, or on `close()`. +Background flushes run on the client's publish executor: a single daemon thread +owned by each client by default. Applications running many clients, or preferring +virtual threads, can supply a shared executor via `publishExecutor` (see the +options table); it is never shut down by the client. + ## Subscribing There are two subscriber APIs over one shared poll engine: a non-blocking @@ -196,6 +201,7 @@ unrecoverable poll failure is rethrown from `hasNext()`. | `maxRetryDuration` | 10m | Max time to retry a failed flush before `FlushTimeoutException`. Must be < the workflow's publisher TTL (15m) to preserve exactly-once delivery | | `payloadConverters` | standard set | Per-item serialization. Payload conversion only — the client's codec chain runs once on the envelope, never per item | | `pollExecutor` | 2 daemon threads, client-owned | Scheduler shared by the client's subscriptions. It runs the short update-admission and delivery steps and poll cooldowns — never held during the long poll itself. A user-supplied executor is never shut down by the client; supply a bigger pool for many subscriptions against slow workflows | +| `publishExecutor` | 1 daemon thread, client-owned | Scheduler driving the client's background flushes (periodic ticks and full-buffer/`forceFlush` triggers). A flush occupies a thread while signaling the workflow. A user-supplied executor is never shut down by the client; share one across clients — or use a virtual-thread executor — instead of paying a platform thread per client | | `SubscribeOptions.pollCooldown` | 100ms | Min interval between polls | ## Cross-language protocol diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java index b5348879d4..2bc92e56d1 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java @@ -101,7 +101,8 @@ private WorkflowStreamClient( dataConverter, options.getBatchInterval(), options.getMaxBatchSize(), - options.getMaxRetryDuration()); + options.getMaxRetryDuration(), + options.getPublishExecutor()); } /** @@ -206,7 +207,8 @@ private ScheduledExecutorService pollExecutor() { * *

Also stops this client's live subscriptions (their done futures complete normally, without * {@link WorkflowStreamListener#onCompleted}) and, if the client owns the default poll executor, - * shuts it down. A user-supplied poll executor is never shut down. + * shuts it down. A user-supplied poll or publish executor is never shut down — only this client's + * own tasks on it are stopped. */ @Override public void close() { diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java index ffd0d8e000..5d51ac0d11 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java @@ -24,18 +24,21 @@ public static WorkflowStreamClientOptions getDefaultInstance() { private final Duration maxRetryDuration; private final PayloadConverter[] payloadConverters; @Nullable private final ScheduledExecutorService pollExecutor; + @Nullable private final ScheduledExecutorService publishExecutor; private WorkflowStreamClientOptions( Duration batchInterval, int maxBatchSize, Duration maxRetryDuration, PayloadConverter[] payloadConverters, - @Nullable ScheduledExecutorService pollExecutor) { + @Nullable ScheduledExecutorService pollExecutor, + @Nullable ScheduledExecutorService publishExecutor) { this.batchInterval = batchInterval; this.maxBatchSize = maxBatchSize; this.maxRetryDuration = maxRetryDuration; this.payloadConverters = payloadConverters.clone(); this.pollExecutor = pollExecutor; + this.publishExecutor = publishExecutor; } public Duration getBatchInterval() { @@ -59,12 +62,18 @@ public ScheduledExecutorService getPollExecutor() { return pollExecutor; } + @Nullable + public ScheduledExecutorService getPublishExecutor() { + return publishExecutor; + } + public static final class Builder { private Duration batchInterval = WorkflowStreamConstants.DEFAULT_BATCH_INTERVAL; private int maxBatchSize; private Duration maxRetryDuration = WorkflowStreamConstants.DEFAULT_MAX_RETRY_DURATION; private PayloadConverter[] payloadConverters = new PayloadConverter[0]; @Nullable private ScheduledExecutorService pollExecutor; + @Nullable private ScheduledExecutorService publishExecutor; private Builder() {} @@ -128,9 +137,30 @@ public Builder setPollExecutor(ScheduledExecutorService pollExecutor) { return this; } + /** + * Executor that drives the client's background publish path: the periodic flushes, and the + * flushes triggered by a full buffer or {@code forceFlush}. The caller owns its lifecycle; it + * is shared across all publishes of this client and must have at least one thread. A flush + * blocks while signaling the workflow, so it occupies an executor thread for the duration of + * each send — supply a pool sized for the number of clients that may flush concurrently, or a + * virtual-thread executor to make that cost negligible. + * + *

Default: a single-thread daemon executor created lazily and owned by the client's + * publisher (shut down by {@link WorkflowStreamClient#close}). + */ + public Builder setPublishExecutor(ScheduledExecutorService publishExecutor) { + this.publishExecutor = publishExecutor; + return this; + } + public WorkflowStreamClientOptions build() { return new WorkflowStreamClientOptions( - batchInterval, maxBatchSize, maxRetryDuration, payloadConverters, pollExecutor); + batchInterval, + maxBatchSize, + maxRetryDuration, + payloadConverters, + pollExecutor, + publishExecutor); } } } diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java index 88c9270a7a..29f50a7063 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java @@ -9,6 +9,12 @@ import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflow; import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflowImpl; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.junit.Assert; import org.junit.Rule; import org.junit.Test; @@ -204,4 +210,55 @@ public void testCloseStopsIteration() { stub.signal("finish"); stub.getResult(Void.class); } + + /** + * Two clients share one user-supplied publish executor. Neither calls flush: the background flush + * loop running on the shared executor must deliver both items, and closing both clients must + * leave the executor — which the caller owns — running. + */ + @Test + public void testUserPublishExecutorSharedAcrossClients() throws Exception { + WorkflowStub stub = startHostWorkflow(); + ScheduledExecutorService publishExecutor = + Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r, "user-publish-executor"); + t.setDaemon(true); + return t; + }); + WorkflowStreamClientOptions options = + WorkflowStreamClientOptions.newBuilder() + .setBatchInterval(Duration.ofMillis(100)) + .setPublishExecutor(publishExecutor) + .build(); + String workflowId = stub.getExecution().getWorkflowId(); + try (WorkflowStreamClient clientA = + WorkflowStreamClient.newInstance( + testWorkflowRule.getWorkflowClient(), workflowId, options); + WorkflowStreamClient clientB = + WorkflowStreamClient.newInstance( + testWorkflowRule.getWorkflowClient(), workflowId, options)) { + clientA.topic("evt").publish("a", false); + clientB.topic("evt").publish("b", false); + + // The two batches may arrive in either order; both must arrive. + try (WorkflowStreamSubscription subscription = clientA.subscribe(FAST_POLL)) { + List values = new ArrayList<>(); + values.add(decode(subscription.next())); + values.add(decode(subscription.next())); + Assert.assertEquals(2, values.size()); + Assert.assertTrue(values.contains("a")); + Assert.assertTrue(values.contains("b")); + } + } + + Assert.assertFalse(publishExecutor.isShutdown()); + CountDownLatch ran = new CountDownLatch(1); + publishExecutor.execute(ran::countDown); + Assert.assertTrue(ran.await(10, TimeUnit.SECONDS)); + publishExecutor.shutdownNow(); + + stub.signal("finish"); + stub.getResult(Void.class); + } } From 02042ef45e5d3efb61988ccf6dc4d689c0d0d83f Mon Sep 17 00:00:00 2001 From: Di Wu Date: Thu, 20 Aug 2026 15:43:52 -0700 Subject: [PATCH 3/3] Drop virtual-thread references from publishExecutor docs 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. --- contrib/temporal-workflowstreams/README.md | 8 ++++---- .../workflowstreams/WorkflowStreamClientOptions.java | 3 +-- .../workflowstreams/internal/StreamPublisher.java | 8 ++++---- 3 files changed, 9 insertions(+), 10 deletions(-) diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index b71404d5b0..70702e6d3b 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -113,9 +113,9 @@ when the buffer reaches the max batch size, on `forceFlush`, on an explicit `flush()`, or on `close()`. Background flushes run on the client's publish executor: a single daemon thread -owned by each client by default. Applications running many clients, or preferring -virtual threads, can supply a shared executor via `publishExecutor` (see the -options table); it is never shut down by the client. +owned by each client by default. Applications running many clients can supply a +shared executor via `publishExecutor` (see the options table); it is never shut +down by the client. ## Subscribing @@ -201,7 +201,7 @@ unrecoverable poll failure is rethrown from `hasNext()`. | `maxRetryDuration` | 10m | Max time to retry a failed flush before `FlushTimeoutException`. Must be < the workflow's publisher TTL (15m) to preserve exactly-once delivery | | `payloadConverters` | standard set | Per-item serialization. Payload conversion only — the client's codec chain runs once on the envelope, never per item | | `pollExecutor` | 2 daemon threads, client-owned | Scheduler shared by the client's subscriptions. It runs the short update-admission and delivery steps and poll cooldowns — never held during the long poll itself. A user-supplied executor is never shut down by the client; supply a bigger pool for many subscriptions against slow workflows | -| `publishExecutor` | 1 daemon thread, client-owned | Scheduler driving the client's background flushes (periodic ticks and full-buffer/`forceFlush` triggers). A flush occupies a thread while signaling the workflow. A user-supplied executor is never shut down by the client; share one across clients — or use a virtual-thread executor — instead of paying a platform thread per client | +| `publishExecutor` | 1 daemon thread, client-owned | Scheduler driving the client's background flushes (periodic ticks and full-buffer/`forceFlush` triggers). A flush occupies a thread while signaling the workflow. A user-supplied executor is never shut down by the client; share one across clients instead of paying a thread per client | | `SubscribeOptions.pollCooldown` | 100ms | Min interval between polls | ## Cross-language protocol diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java index 5d51ac0d11..2f93696686 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java @@ -142,8 +142,7 @@ public Builder setPollExecutor(ScheduledExecutorService pollExecutor) { * flushes triggered by a full buffer or {@code forceFlush}. The caller owns its lifecycle; it * is shared across all publishes of this client and must have at least one thread. A flush * blocks while signaling the workflow, so it occupies an executor thread for the duration of - * each send — supply a pool sized for the number of clients that may flush concurrently, or a - * virtual-thread executor to make that cost negligible. + * each send — supply a pool sized for the number of clients that may flush concurrently. * *

Default: a single-thread daemon executor created lazily and owned by the client's * publisher (shut down by {@link WorkflowStreamClient#close}). diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java index 3498ad4105..7c3d723cb3 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java @@ -72,10 +72,10 @@ public StreamPublisher( /** * @param executor drives the background flush loop (the periodic ticks and the flushes triggered * by a full buffer or {@code forceFlush}). When non-null the caller owns its lifecycle and it - * is never shut down by this publisher, so many publishers can share one executor (e.g. a - * virtual-thread executor); when null a single-thread executor is created lazily, owned by - * this publisher, and shut down by {@link #close}. Flushes block while signaling the - * workflow, so each in-flight flush occupies an executor thread for the duration of the send. + * is never shut down by this publisher, so many publishers can share one executor; when null + * a single-thread executor is created lazily, owned by this publisher, and shut down by + * {@link #close}. Flushes block while signaling the workflow, so each in-flight flush + * occupies an executor thread for the duration of the send. */ public StreamPublisher( SignalFunction signal,