From 246063760c63cad80f79206333e4ad60050487d2 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Thu, 20 Aug 2026 15:29:55 -0700 Subject: [PATCH 1/6] 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 ee0ce9478386a7597c48da2fd116bb25eb5e8094 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Thu, 20 Aug 2026 15:32:33 -0700 Subject: [PATCH 2/6] 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 d5f8944f38faa714dbcaed0d9f71edf021cb8285 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Thu, 20 Aug 2026 15:43:52 -0700 Subject: [PATCH 3/6] 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, From a442a198a190254f2c0197aff493400d62cbdfca Mon Sep 17 00:00:00 2001 From: Di Wu Date: Mon, 24 Aug 2026 20:41:53 -0700 Subject: [PATCH 4/6] Gate every background flush after a flush timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- contrib/temporal-workflowstreams/README.md | 5 ++ .../internal/StreamPublisher.java | 35 ++++++++- .../workflowstreams/StreamPublisherTest.java | 78 +++++++++++++++++-- 3 files changed, 108 insertions(+), 10 deletions(-) diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index 70702e6d3b..88e3fb4387 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -117,6 +117,11 @@ 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. +If a flush retry exceeds `maxRetryDuration`, background flushing stops for that +client — neither the periodic tick nor a `forceFlush`/max-batch-size trigger +sends again. Later items stay buffered until an explicit `flush()` or `close()`, +which rethrows the `FlushTimeoutException` for the dropped batch. + ## Subscribing There are two subscriber APIs over one shared poll engine: a non-blocking 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 7c3d723cb3..faf4309ac1 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 @@ -10,6 +10,7 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -19,7 +20,9 @@ * Owns the client-side publish path: it buffers published values, batches them, and sends each * batch to the workflow via the injected signal function. It assigns the per-publisher dedup key (a * stable publisher ID plus a monotonic sequence advanced only on a confirmed send) so the workflow - * can drop duplicates, and it retries a failed batch until the max retry duration elapses. + * can drop duplicates, and it retries a failed batch until the max retry duration elapses. Once a + * background flush exceeds that duration the background loop stops for good and the resulting + * {@link FlushTimeoutException} is deferred to the next {@link #flush} or {@link #close}. * *

The signal function is injected (rather than holding a client) so the publish path can be * exercised in isolation. Internal to the workflow streams module. @@ -49,6 +52,9 @@ public interface SignalFunction { private long pendingStartNanos; private boolean started; private boolean closed; + // Set when a background flush timed out: the loop is stopped for good, so no background send may + // run before flush()/close() surfaces the deferred error. Guarded by stateLock. + private boolean loopStopped; private FlushTimeoutException deferredError; // The executor driving the flush loop once started; the owned one when no user executor was // supplied. Guarded by stateLock. @@ -101,6 +107,13 @@ public StreamPublisher( * publish} call itself instead of poisoning the buffer and silently wedging every later item * behind it in the background flush loop. * + *

After a background flush exceeds the max retry duration the background loop is stopped + * permanently — neither the periodic tick nor a {@code forceFlush}/max-batch-size trigger sends + * again. Items published afterwards stay buffered until {@link #flush} or {@link #close} drains + * them, and that call surfaces the deferred {@link FlushTimeoutException} first (flush) or after + * the final drain (close). This keeps a caller-owned executor untouched without letting more data + * ship before the failure is reported. + * * @throws RuntimeException if no configured payload converter accepts {@code value} */ public void publish(String topic, Object value, boolean forceFlush) { @@ -109,14 +122,19 @@ public void publish(String topic, Object value, boolean forceFlush) { ScheduledExecutorService toTrigger = null; synchronized (stateLock) { buffer.add(entry); - trigger = forceFlush || (maxBatchSize > 0 && buffer.size() >= maxBatchSize); + trigger = (forceFlush || (maxBatchSize > 0 && buffer.size() >= maxBatchSize)) && !loopStopped; if (!closed) { ensureStartedLocked(); toTrigger = scheduler; } } if (trigger && toTrigger != null) { - toTrigger.execute(this::backgroundFlush); + try { + toTrigger.execute(this::backgroundFlush); + } catch (RejectedExecutionException e) { + // The executor stopped between reading it and submitting (close(), or a user executor + // shut down by its owner). The item stays buffered for flush()/close() to drain. + } } } @@ -142,15 +160,24 @@ private void ensureStartedLocked() { } private void backgroundFlush() { + synchronized (stateLock) { + if (loopStopped) { + // A timed-out flush stopped the loop; a task queued before that must not send. + return; + } + } try { doFlush(); } 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. + // flush/close surface it and stop the loop for good: with a user-supplied executor + // cancelling the periodic task is not enough, since publish() can still submit a + // triggered flush onto the still-live executor. ScheduledFuture toCancel; ScheduledExecutorService toStop; synchronized (stateLock) { deferredError = e; + loopStopped = true; toCancel = flushTask; toStop = ownedSchedulerLocked(); } 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 04eeb58ee2..d9bf6d148b 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 @@ -308,12 +308,14 @@ public void testUserExecutorDrivesFlushesAndSurvivesClose() throws InterruptedEx } /** - * 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. + * After a background flush exceeds the max retry duration, every background send is gated: the + * periodic task on a user executor is cancelled (not the executor) and a {@code forceFlush} + * trigger no longer submits. 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 { + public void testTriggeredFlushGatedAfterFlushTimeoutWithUserExecutor() + throws InterruptedException { RecordingSignal signal = new RecordingSignal(); signal.failure = new RuntimeException("boom"); ScheduledExecutorService user = newNamedExecutor("user-publish-executor"); @@ -327,8 +329,9 @@ public void testUserExecutorTaskCancelledAfterFlushTimeout() throws InterruptedE 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. + // forceFlush would otherwise submit onto the still-live user executor and ship "b" + // before anyone has seen the deferred timeout. + publisher.publish("t", "b", true); Thread.sleep(300); Assert.assertTrue(signal.recorded().isEmpty()); @@ -344,4 +347,67 @@ public void testUserExecutorTaskCancelledAfterFlushTimeout() throws InterruptedE assertExecutorStillRunsTasks(user); user.shutdownNow(); } + + /** The max-batch-size trigger is gated after a flush timeout just like {@code forceFlush}. */ + @Test + public void testMaxBatchSizeTriggerGatedAfterFlushTimeout() 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), 2, Duration.ofMillis(1), user); + + publisher.publish("t", "a", false); // below maxBatchSize: only a tick can send it + eventually(Duration.ofSeconds(5), () -> Assert.assertEquals(1, signal.attempts())); + Thread.sleep(300); + + signal.failure = null; + publisher.publish("t", "b", false); + publisher.publish("t", "c", false); // reaches maxBatchSize -> trigger, must be gated + Thread.sleep(300); + Assert.assertTrue(signal.recorded().isEmpty()); + + try { + publisher.close(); + Assert.fail("unreachable"); + } catch (FlushTimeoutException expected) { + } + Assert.assertEquals(1, signal.recorded().size()); + Assert.assertEquals(2, signal.recorded().get(0).items.size()); + Assert.assertEquals("b", decodeItem(signal.recorded().get(0), 0)); + Assert.assertEquals("c", decodeItem(signal.recorded().get(0), 1)); + assertExecutorStillRunsTasks(user); + user.shutdownNow(); + } + + /** + * The client-owned executor behaves identically: after a flush timeout a triggered publish + * neither sends nor fails the caller with the executor's own RejectedExecutionException — the + * deferred timeout is what surfaces, from close(). + */ + @Test + public void testTriggeredFlushGatedAfterFlushTimeoutWithOwnedExecutor() + throws InterruptedException { + RecordingSignal signal = new RecordingSignal(); + signal.failure = new RuntimeException("boom"); + StreamPublisher publisher = + newPublisher(signal, Duration.ofMillis(20), 0, Duration.ofMillis(1)); + + publisher.publish("t", "a", false); + eventually(Duration.ofSeconds(5), () -> Assert.assertEquals(1, signal.attempts())); + Thread.sleep(300); + + signal.failure = null; + publisher.publish("t", "b", true); // must not hit the shut-down owned executor + Thread.sleep(300); + Assert.assertTrue(signal.recorded().isEmpty()); + + try { + publisher.close(); + Assert.fail("unreachable"); + } catch (FlushTimeoutException expected) { + } + Assert.assertEquals(1, signal.recorded().size()); + Assert.assertEquals("b", decodeItem(signal.recorded().get(0), 0)); + } } From 9ec31858e33cc2534542ad1662d0686b6e9ad8d8 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Mon, 24 Aug 2026 21:02:18 -0700 Subject: [PATCH 5/6] Keep client teardown and queued flushes safe around a failing flush 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) --- contrib/temporal-workflowstreams/README.md | 6 +- .../workflowstreams/WorkflowStreamClient.java | 41 +++++---- .../internal/StreamPublisher.java | 18 ++-- .../workflowstreams/StreamPublisherTest.java | 84 +++++++++++++++++++ 4 files changed, 125 insertions(+), 24 deletions(-) diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index 88e3fb4387..d7a7448738 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -119,8 +119,10 @@ down by the client. If a flush retry exceeds `maxRetryDuration`, background flushing stops for that client — neither the periodic tick nor a `forceFlush`/max-batch-size trigger -sends again. Later items stay buffered until an explicit `flush()` or `close()`, -which rethrows the `FlushTimeoutException` for the dropped batch. +sends again. Later items stay buffered until an explicit `flush()` or `close()` +drains them. The two report the dropped batch at different points: `flush()` +rethrows the `FlushTimeoutException` before sending anything (call it again to +drain), while `close()` drains first and rethrows afterwards. ## Subscribing 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 2bc92e56d1..26f563f0db 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 @@ -207,28 +207,35 @@ 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 or publish executor is never shut down — only this client's - * own tasks on it are stopped. + * shuts it down. That teardown runs even when the final flush fails, so a thrown {@link + * FlushTimeoutException} never leaves subscriptions polling. 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() { - publisher.close(); - for (SubscriptionDriver driver : liveSubscriptions.toArray(new SubscriptionDriver[0])) { - driver.close(); - } - ScheduledExecutorService owned; - synchronized (this) { - owned = ownedPollExecutor; - } - if (owned != null) { - owned.shutdown(); - try { - if (!owned.awaitTermination(1, TimeUnit.SECONDS)) { + // The final flush can throw (a deferred FlushTimeoutException, or a failing signal), and the + // rest of the teardown must still run: otherwise live subscriptions keep polling forever and + // the owned poll executor is never shut down. + try { + publisher.close(); + } finally { + for (SubscriptionDriver driver : liveSubscriptions.toArray(new SubscriptionDriver[0])) { + driver.close(); + } + ScheduledExecutorService owned; + synchronized (this) { + owned = ownedPollExecutor; + } + if (owned != null) { + owned.shutdown(); + try { + if (!owned.awaitTermination(1, TimeUnit.SECONDS)) { + owned.shutdownNow(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); owned.shutdownNow(); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - owned.shutdownNow(); } } } 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 faf4309ac1..5198ab2943 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 @@ -154,15 +154,23 @@ private void ensureStartedLocked() { return t; }); } - flushTask = - scheduler.scheduleWithFixedDelay( - this::backgroundFlush, batchIntervalMs, batchIntervalMs, TimeUnit.MILLISECONDS); + try { + flushTask = + scheduler.scheduleWithFixedDelay( + this::backgroundFlush, batchIntervalMs, batchIntervalMs, TimeUnit.MILLISECONDS); + } catch (RejectedExecutionException e) { + // A user-supplied executor was already shut down. Don't fail the publish call with the + // executor's own exception: items stay buffered for flush()/close() to drain on the + // caller's thread, as they do after a flush timeout stops the loop. + } } private void backgroundFlush() { synchronized (stateLock) { - if (loopStopped) { - // A timed-out flush stopped the loop; a task queued before that must not send. + if (loopStopped || closed) { + // The loop is stopped (a timed-out flush) or the publisher is closed. A task already + // queued at that point must not send: with a user-supplied executor nothing purges the + // queue, so this is the only thing keeping a flush from running after close() returned. return; } } 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 d9bf6d148b..b8aa05c285 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 @@ -410,4 +410,88 @@ public void testTriggeredFlushGatedAfterFlushTimeoutWithOwnedExecutor() Assert.assertEquals(1, signal.recorded().size()); Assert.assertEquals("b", decodeItem(signal.recorded().get(0), 0)); } + + /** + * A triggered flush can still be queued on a shared user executor when close() returns — the + * caller owns that executor, so nothing purges its queue. The queued task must not signal after + * close() has returned and the caller believes the publisher is quiesced. + */ + @Test + public void testQueuedFlushDoesNotSendAfterClose() throws InterruptedException { + RecordingSignal signal = new RecordingSignal(); + signal.failure = new RuntimeException("boom"); + ScheduledExecutorService user = newNamedExecutor("user-publish-executor"); + CountDownLatch blocking = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + user.execute( + () -> { + blocking.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + Assert.assertTrue(blocking.await(5, TimeUnit.SECONDS)); + + StreamPublisher publisher = + new StreamPublisher( + signal, + DC, + Duration.ofHours(1), + 0, + WorkflowStreamConstants.DEFAULT_MAX_RETRY_DURATION, + user); + publisher.publish("t", "a", true); // the triggered flush queues behind the blocked task + + // close() drains on the caller thread; the transient failure propagates and leaves the batch + // pending, which is exactly what the queued flush would retry. + try { + publisher.close(); + Assert.fail("unreachable"); + } catch (RuntimeException e) { + Assert.assertEquals("boom", e.getMessage()); + } + + signal.failure = null; + release.countDown(); + Thread.sleep(200); + Assert.assertEquals("no send may run after close() returned", 1, signal.attempts()); + Assert.assertTrue(signal.recorded().isEmpty()); + assertExecutorStillRunsTasks(user); + user.shutdownNow(); + } + + /** + * A user executor its owner already shut down must not surface the executor's own + * RejectedExecutionException from publish() — neither when scheduling the periodic tick nor when + * submitting a triggered flush. Items stay buffered and an explicit flush still drains them on + * the caller's thread. + */ + @Test + public void testPublishToleratesShutDownUserExecutor() { + RecordingSignal signal = new RecordingSignal(); + ScheduledExecutorService user = newNamedExecutor("user-publish-executor"); + user.shutdownNow(); + + StreamPublisher publisher = + new StreamPublisher( + signal, + DC, + Duration.ofMillis(20), + 0, + WorkflowStreamConstants.DEFAULT_MAX_RETRY_DURATION, + user); + publisher.publish("t", "a", true); // both the schedule and the trigger are rejected + publisher.publish("t", "b", false); + Assert.assertTrue(signal.recorded().isEmpty()); + + publisher.flush(); + Assert.assertEquals(1, signal.recorded().size()); + Assert.assertEquals(2, signal.recorded().get(0).items.size()); + Assert.assertEquals("a", decodeItem(signal.recorded().get(0), 0)); + Assert.assertEquals("b", decodeItem(signal.recorded().get(0), 1)); + + publisher.close(); + } } From 32e3d8e09ec34bb49d1b5cdbb9741cc87fcdce64 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Mon, 24 Aug 2026 21:49:55 -0700 Subject: [PATCH 6/6] Extract the client teardown steps out of close() 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) --- .../workflowstreams/WorkflowStreamClient.java | 43 +++++++++++-------- 1 file changed, 26 insertions(+), 17 deletions(-) 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 26f563f0db..f18e021cfa 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 @@ -219,24 +219,33 @@ public void close() { try { publisher.close(); } finally { - for (SubscriptionDriver driver : liveSubscriptions.toArray(new SubscriptionDriver[0])) { - driver.close(); - } - ScheduledExecutorService owned; - synchronized (this) { - owned = ownedPollExecutor; - } - if (owned != null) { - owned.shutdown(); - try { - if (!owned.awaitTermination(1, TimeUnit.SECONDS)) { - owned.shutdownNow(); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - owned.shutdownNow(); - } + stopSubscriptions(); + shutdownOwnedPollExecutor(); + } + } + + private void stopSubscriptions() { + for (SubscriptionDriver driver : liveSubscriptions.toArray(new SubscriptionDriver[0])) { + driver.close(); + } + } + + private void shutdownOwnedPollExecutor() { + ScheduledExecutorService owned; + synchronized (this) { + owned = ownedPollExecutor; + } + if (owned == null) { + return; + } + owned.shutdown(); + try { + if (!owned.awaitTermination(1, TimeUnit.SECONDS)) { + owned.shutdownNow(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + owned.shutdownNow(); } }