diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index ad630c89b7..d7a7448738 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -112,6 +112,18 @@ 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 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()` +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 There are two subscriber APIs over one shared poll engine: a non-blocking @@ -196,6 +208,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 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/WorkflowStreamClient.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java index b5348879d4..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 @@ -101,7 +101,8 @@ private WorkflowStreamClient( dataConverter, options.getBatchInterval(), options.getMaxBatchSize(), - options.getMaxRetryDuration()); + options.getMaxRetryDuration(), + options.getPublishExecutor()); } /** @@ -206,28 +207,45 @@ 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. 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(); + // 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 { + 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) { - owned.shutdown(); - try { - if (!owned.awaitTermination(1, TimeUnit.SECONDS)) { - owned.shutdownNow(); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); + if (owned == null) { + return; + } + owned.shutdown(); + try { + if (!owned.awaitTermination(1, TimeUnit.SECONDS)) { owned.shutdownNow(); } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + owned.shutdownNow(); } } 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..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 @@ -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,29 @@ 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. + * + *

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/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/internal/StreamPublisher.java index f07e7095ad..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 @@ -10,14 +10,19 @@ 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; +import javax.annotation.Nullable; /** * 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. @@ -36,6 +41,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<>(); @@ -45,8 +52,16 @@ 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. 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 +72,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; 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; } /** @@ -73,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) { @@ -81,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. + } } } @@ -97,27 +143,54 @@ 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; + }); + } + 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 || 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; + } + } 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; - toStop = scheduler; + loopStopped = true; + toCancel = flushTask; + toStop = ownedSchedulerLocked(); + } + if (toCancel != null) { + toCancel.cancel(false); } if (toStop != null) { toStop.shutdown(); @@ -226,18 +299,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 +338,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..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 @@ -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,226 @@ 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, 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 testTriggeredFlushGatedAfterFlushTimeoutWithUserExecutor() + 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; + // 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()); + + 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(); + } + + /** 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)); + } + + /** + * 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(); + } } 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); + } }