diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md
index ad630c89b7..70702e6d3b 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 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 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..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() {
    *
    * <p>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..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.
+     *
+     * <p>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..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
@@ -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<PublishEntry> 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; 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<PublishInput> signals = new ArrayList<>();
+    final List<String> 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<PublishInput> recorded() {
       return new ArrayList<>(signals);
     }
+
+    synchronized List<String> 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();
+  }
 }
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<String> 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);
+  }
 }