# Workflow Streams - Java SDK

> Stream events from a Workflow to subscribers using the Temporal Java SDK Workflow Streams contrib module.

> **Public Preview**

[Workflow Streams](/workflow-streams) adds a durable event channel to a Workflow, letting outside observers follow its progress in real time. This page walks through enabling a stream, publishing events from Workflows and Activities, subscribing to a stream with either the blocking iterator or the non-blocking listener, and keeping a stream running across long-lived Workflows.

## Enable streaming on a Workflow

The library ships as the `io.temporal:temporal-workflowstreams` contrib module. Enable streaming by constructing a `WorkflowStream` once via `WorkflowStream.newInstance()`, preferably in a [`@WorkflowInit`](https://www.javadoc.io/doc/io.temporal/temporal-sdk/latest/io/temporal/workflow/WorkflowInit.html) constructor. The factory registers the stream's handlers on the current Workflow, and a `@WorkflowInit` constructor runs before any handler dispatch, so polls and offset Queries arriving with the first Workflow Task (for example, from Update-With-Start) are accepted rather than rejected.

```java
import io.temporal.workflow.WorkflowInit;
import io.temporal.workflowstreams.WorkflowStream;
import io.temporal.workflowstreams.WorkflowStreamState;

public class OrderInput {
  public String orderId;
  public WorkflowStreamState streamState;
}

public class OrderWorkflowImpl implements OrderWorkflow {
  private final WorkflowStream stream;

  @WorkflowInit
  public OrderWorkflowImpl(OrderInput input) {
    stream = WorkflowStream.newInstance(input.streamState);
  }

  @Override
  public void execute(OrderInput input) {
    // ... rest of the workflow
  }
}
```

`WorkflowStream.newInstance` creates the in-memory event log and registers the publish Signal, subscribe Update, and offset Query handlers on the current Workflow. The `priorState` argument may be `null` and is only needed for Continue-As-New rollovers: pass `null` (or call the no-argument overload) on a fresh start, and the carried `WorkflowStreamState` after a rollover (see [Stream from long-running Workflows](#continue-as-new)).

Constructing the stream at the top of the Workflow method also works — Signals received earlier are buffered by the SDK — but polls and offset Queries are rejected until the stream exists, so prefer `@WorkflowInit`. Construct exactly one `WorkflowStream` per Workflow.

## Publish from a Workflow

Bind a [topic name](/workflow-streams#topics) with `stream.topic(name)`, then call `publish()` on the returned `WorkflowTopicHandle` to append events. Repeated calls with the same name return the same handle.

```java
import io.temporal.activity.ActivityOptions;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInit;
import io.temporal.workflowstreams.WorkflowStream;
import io.temporal.workflowstreams.WorkflowTopicHandle;
import java.time.Duration;

public class StatusEvent {
  public String state;
  public int progress;
  public String detail;

  public StatusEvent() {}

  public StatusEvent(String state, int progress, String detail) {
    this.state = state;
    this.progress = progress;
    this.detail = detail;
  }
}

public class OrderWorkflowImpl implements OrderWorkflow {
  private final WorkflowStream stream;
  private final WorkflowTopicHandle status;

  private final OrderActivities activities =
      Workflow.newActivityStub(
          OrderActivities.class,
          ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofMinutes(1)).build());

  @WorkflowInit
  public OrderWorkflowImpl(OrderInput input) {
    stream = WorkflowStream.newInstance(input.streamState);
    status = stream.topic("status");
  }

  @Override
  public void execute(OrderInput input) {
    status.publish(new StatusEvent("validating", 0, "checking inventory"));
    activities.validateOrder(input.orderId);

    status.publish(new StatusEvent("charging", 33, "authorizing payment"));
    activities.chargePayment(input.orderId);

    status.publish(new StatusEvent("shipping", 66, "dispatching to warehouse"));
    activities.dispatchOrder(input.orderId);

    status.publish(new StatusEvent("completed", 100, ""));
  }
}
```

`publish()` runs the payload converter to encode each value. The codec chain (encryption, compression, etc.) runs once on the Signal or Update envelope that carries the batch, never per item, so encryption and compression are applied exactly once each direction.

Unlike the Python and TypeScript SDKs, Java topics carry no per-topic type binding. A topic handle is bound only to a name; published values are `Object` and subscribers decode each item from its raw payload (see [Subscribe](#subscribe)). To customize per-item serialization, pass `WorkflowStreamOptions.newBuilder().setPayloadConverters(...)` to `WorkflowStream.newInstance`, and use the matching converters on the subscriber side. There is no public accessor for the Worker's configured data converter inside Workflow code, so a custom converter can't be picked up automatically; pass matching payload converters explicitly to keep Workflow-side publishes consistent with the rest of your Workflow. A pre-built `io.temporal.api.common.v1.Payload` may also be passed to `publish()`, bypassing conversion.

## Publish from a client

Any process that has a Temporal Client and the target Workflow Id can publish to that Workflow's stream by constructing a `WorkflowStreamClient`. This is the general pattern and covers HTTP backends, starters, one-off scripts, other Workflows' Activities, and standalone Activities.

Construct one with:

```java
WorkflowStreamClient.newInstance(workflowClient, workflowId)
```

Then use it the same way you would the Workflow-side handle: bind a topic, publish through it, and let try-with-resources flush on scope exit (`close()` guarantees a final flush of buffered items).

When events originate in an Activity, publish from the Activity directly rather than returning them for the Workflow to forward. The Workflow hosts the stream but doesn't read its own stream; it processes the Activity's return value and emits its own lifecycle events. Keeping Workflow state independent of streamed output is what lets retried Activity attempts surface to subscribers without polluting the Workflow's durable state. See [How events are delivered](/workflow-streams#how-events-are-delivered).

```java
import io.temporal.client.WorkflowClient;
import io.temporal.workflowstreams.TopicHandle;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamClientOptions;
import java.time.Duration;

public void publishStatus(WorkflowClient workflowClient, String workflowId) {
  WorkflowStreamClientOptions options =
      WorkflowStreamClientOptions.newBuilder()
          .setBatchInterval(Duration.ofMillis(200))
          .build();
  try (WorkflowStreamClient streamClient =
      WorkflowStreamClient.newInstance(workflowClient, workflowId, options)) {
    TopicHandle status = streamClient.topic("status");
    status.publish(new StatusEvent("started", 0, ""));
    // ...
    // Buffer is flushed automatically on close().
  }
}
```

Inside an Activity scheduled by a Workflow, `WorkflowStreamClient.fromActivity()` infers the Temporal Client and the parent Workflow Id from the Activity context, so you don't have to thread them through the Activity's input:

```java
import io.temporal.activity.Activity;
import io.temporal.workflowstreams.TopicHandle;
import io.temporal.workflowstreams.WorkflowStreamClient;

public class Delta {
  public String text;
}

public void streamDeltas(String orderId) {
  try (WorkflowStreamClient streamClient = WorkflowStreamClient.fromActivity()) {
    TopicHandle deltas = streamClient.topic("delta");
    for (Delta delta : generateDeltas(orderId)) {
      deltas.publish(delta);
      Activity.getExecutionContext().heartbeat(null);
    }
    // Buffer is flushed automatically on close().
  }
}
```

For a [standalone Activity](/develop/java/activities/standalone-activities) (one started directly via the Client rather than from a Workflow), there is no parent Workflow context to infer, so `fromActivity()` throws an `IllegalStateException`. Fall back to the general pattern with `Activity.getExecutionContext().getWorkflowClient()` and the target Workflow Id threaded through the Activity's input.

Two operations give the application explicit control over when batches ship: the `forceFlush` argument on a publish for latency, and `client.flush()` for confirmation that prior publications have landed.

Pass `true` as the `forceFlush` argument on a publish to wake the background flusher so the current buffer ships without waiting for the next interval. The flusher only runs while the client is open (between construction and `close()`). The call returns immediately after appending to the buffer and signaling the flusher. It doesn't wait for delivery to the Workflow or to subscribers:

```java
deltas.publish(delta, /* forceFlush */ true);
```

Use it for latency-sensitive events: the first delta of a response so the user sees something fast, or punctuated events like `RETRY` and `STATUS_CHANGE`. See [Tuning](/workflow-streams#tuning) for the trade-off against history pressure.

Use `client.flush()` when you need a mid-stream barrier. Successful completion of the flush is proof that the Temporal server has received all prior publications, so subsequent work that depends on those events being durable can proceed. The client stays open for further publishing afterward. `close()` already flushes on its way out, so the explicit call is only for barriers in the middle:

```java
try (WorkflowStreamClient streamClient =
    WorkflowStreamClient.newInstance(workflowClient, workflowId)) {
  TopicHandle deltas = streamClient.topic("delta");

  for (Delta delta : firstPhase()) {
    deltas.publish(delta);
  }

  streamClient.flush();
  String checkpointId = recordPhaseOneComplete(); // only safe once phase-one events are durable

  for (Delta delta : secondPhase(checkpointId)) {
    deltas.publish(delta);
  }
}
```

Batches also ship early once the buffer reaches `maxBatchSize`, if you set one on `WorkflowStreamClientOptions`; by default only the interval, `forceFlush`, `flush()`, and `close()` trigger a flush.

`publish()` is non-blocking and applies no backpressure. From an Activity or other client, it appends to the client's in-memory buffer and returns (the value goes through the payload converters immediately, so an unconvertible value fails the `publish()` call rather than a later background flush). From inside a Workflow, it appends synchronously to the in-memory log. [Subscribers](/workflow-streams#subscribing) pull from the Workflow's log on their own schedule, so a slow subscriber doesn't slow down [publishers](/workflow-streams#publishing). If a publisher emits faster than batches can ship to the server, the buffer grows: the process uses more memory, the stream falls further behind real time, and at the limit Signals can't keep up.

If your application needs to bound this (to cap memory, to keep the stream close to real time, or to apply a policy when the publisher overruns the network), apply that policy upstream of `publish()`. The choice (block, drop, error, sample) is application-specific, and Workflow Streams doesn't pick one for you.

## Subscribe

[Subscribing](/workflow-streams#subscribing) uses the same client construction as publishing: `WorkflowStreamClient.newInstance(workflowClient, workflowId)` from any process that has a Temporal Client, or `fromActivity()` inside an Activity. Subscribing from an Activity is less common in practice, so the general client case is the primary example below.

The Java SDK offers two subscriber APIs over one shared poll engine: a blocking iterator for synchronous consumers, and a non-blocking listener that delivers items as callbacks without occupying a caller thread. Neither occupies a thread while a poll is blocked on the server — polling runs on a small executor shared by all of a client's subscriptions (2 daemon threads by default; see `WorkflowStreamClientOptions.Builder.setPollExecutor`) — so many concurrent subscriptions don't mean many threads. Either way, the subscription ends cleanly when the Workflow reaches a terminal state, automatically follows Continue-As-New chains, recovers from Workflow-side log truncation by restarting from the current base offset, handles pagination when a poll response hits the ~1 MB cap, and also ends when the owning `WorkflowStreamClient` is closed.

Items carry the raw `io.temporal.api.common.v1.Payload` in `item.getPayload()`; decode at the call site with your data converter. Offsets are global across all topics, not per-topic.

### Blocking iterator

`client.subscribe(options)` without a listener returns a `WorkflowStreamSubscription`: a blocking, single-use subscription the consuming thread iterates with a for-each loop. The thread blocks waiting for items while polling still runs on the shared executor.

```java
import io.temporal.client.WorkflowClient;
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.workflowstreams.SubscribeOptions;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamItem;
import io.temporal.workflowstreams.WorkflowStreamSubscription;

public void watchOrder(WorkflowClient workflowClient, String orderId) {
  SubscribeOptions options = SubscribeOptions.newBuilder().setTopics("status").build();
  try (WorkflowStreamClient stream = WorkflowStreamClient.newInstance(workflowClient, orderId);
      WorkflowStreamSubscription subscription = stream.subscribe(options)) {
    for (WorkflowStreamItem item : subscription) {
      StatusEvent evt =
          DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
              item.getPayload(), StatusEvent.class, StatusEvent.class);
      System.out.printf("[%3d%%] %s: %s%n", evt.progress, evt.state, evt.detail);
      if (evt.state.equals("completed")) {
        break;
      }
    }
  }
}
```

`SubscribeOptions` controls the subscription: `setTopics` filters by name (unset means all topics), `setFromOffset` resumes from a stored global offset (zero means the beginning), and `setPollCooldown` sets the minimum interval between polls (default 100 milliseconds). A single-topic convenience method, `streamClient.topic("status").subscribe(fromOffset)`, is equivalent to passing one name in `setTopics`.

`subscription.close()` stops the subscription before the next poll; items already fetched still drain. An unrecoverable poll failure is rethrown from `hasNext()`. A subscriber that never publishes doesn't strictly need to close the `WorkflowStreamClient` for flushing (the background flusher only runs for publishers), but closing it releases the poll executor, so keep both in the try-with-resources as above.

### Non-blocking listener

Unique to the Java SDK, the second subscriber API inverts control: instead of parking a thread in an iterator, pass a `WorkflowStreamListener` to `subscribe` and items are delivered as callbacks on the poll executor. `subscribe` returns a `WorkflowStreamSubscriptionHandle` immediately. This is the right shape when one process consumes many streams concurrently — all subscriptions share the client's small executor rather than each pinning a thread.

Callbacks are serialized (never invoked concurrently, with happens-before ordering between invocations) and must not block. The `CompletionStage` returned by `onNext` is the backpressure boundary: return `null` (or an already-completed stage) to receive the next item immediately, or a pending stage to defer both further delivery and the next poll until it completes. A stage that completes exceptionally — or an exception thrown directly from `onNext` — stops the subscription and is reported to `onError`.

```java
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.workflowstreams.SubscribeOptions;
import io.temporal.workflowstreams.WorkflowStreamItem;
import io.temporal.workflowstreams.WorkflowStreamListener;
import io.temporal.workflowstreams.WorkflowStreamSubscriptionHandle;
import java.util.concurrent.CompletionStage;

SubscribeOptions options = SubscribeOptions.newBuilder().setTopics("status").build();
WorkflowStreamSubscriptionHandle handle =
    streamClient.subscribe(
        options,
        new WorkflowStreamListener() {
          @Override
          public CompletionStage<Void> onNext(WorkflowStreamItem item) {
            StatusEvent evt =
                DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
                    item.getPayload(), StatusEvent.class, StatusEvent.class);
            System.out.printf(
                "offset=%d topic=%s state=%s%n", item.getOffset(), item.getTopic(), evt.state);
            return null; // or a pending stage to apply backpressure
          }

          @Override
          public void onCompleted() {
            System.out.println("stream ended");
          }
        });

// The calling thread is free; wait on the handle when you need to.
handle.getDoneFuture().join();
```

`onCompleted` fires once when the stream ends cleanly because the Workflow reached a terminal state. `onError` fires once on an unrecoverable failure (including a failure from `onNext`); its default implementation logs at warn level. `handle.close()` stops the subscription before the next poll without calling `onCompleted`. `handle.getDoneFuture()` completes normally when the stream ends cleanly or the subscription is closed, and exceptionally with the failure passed to `onError`. A single-topic convenience, `streamClient.topic("status").subscribe(fromOffset, listener)`, is also available.

To hand work off a callback without blocking the executor, return a stage that completes when the work is done — for example, `CompletableFuture.runAsync(() -> render(item), renderExecutor)` — and the next item is delivered only after it completes.

### Heterogeneous topics

Every item arrives as a raw `Payload` in `item.getPayload()`, so a single subscription naturally consumes multiple topics whose payload types differ. Pass the topic names to `SubscribeOptions.Builder.setTopics` (or leave it unset for every topic on the stream), dispatch on `item.getTopic()`, and decode into the matching type:

```java
SubscribeOptions options =
    SubscribeOptions.newBuilder().setTopics("status", "progress").build();
try (WorkflowStreamSubscription subscription = stream.subscribe(options)) {
  for (WorkflowStreamItem item : subscription) {
    switch (item.getTopic()) {
      case "status":
        StatusEvent status =
            DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
                item.getPayload(), StatusEvent.class, StatusEvent.class);
        System.out.printf("[status] %s: %s%n", status.state, status.detail);
        break;
      case "progress":
        ProgressEvent progress =
            DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
                item.getPayload(), ProgressEvent.class, ProgressEvent.class);
        System.out.printf("[progress] %s%n", progress.message);
        break;
    }
  }
}
```

A single subscription over multiple topics also avoids the cancellation race that two concurrent subscribers would create. Because `item.getPayload()` is the raw payload, it's also the right shape when you want to forward the bytes through to another system without decoding them.

### Closing the stream 

A subscriber's loop or listener doesn't know when the publisher is done. How you [close a stream](/workflow-streams#closing-the-stream) depends on what the application needs. As one example, a common pattern combines two pieces:

1. **An in-band terminator.** The Workflow or its Activity publishes a sentinel event the subscriber recognizes and breaks on. In the `watchOrder` example above, `new StatusEvent("completed", 100, "")` is the minimal form, and the consumer's `if (evt.state.equals("completed")) break` is the matching half. Each subscription decides what its own end-of-stream marker is.
2. **A brief overlap before the Workflow returns.** A poll Update that is still in flight when the Workflow returns is consumed by the subscription transparently, and no new polls can complete after that. If the Workflow returns immediately after publishing the terminator, subscribers may miss it.

There are two ways to provide that overlap.

- [Fixed sleep](/workflow-streams#fixed-sleep). Sleep between the terminator and the return so any in-flight poll has time to fetch the terminator before the Workflow exits:

  ```java
  // at the end of the workflow method
  status.publish(new StatusEvent("completed", 100, ""));
  Workflow.sleep(Duration.ofSeconds(30));
  return result;
  ```

- [Acknowledgment handshake](/workflow-streams#acknowledgment-handshake). The subscriber sends a Signal once it has the terminator; the Workflow waits up to a timeout, returning as soon as the ack arrives:

  ```java
  public class ChatWorkflowImpl implements ChatWorkflow {
    private boolean subscriberDone = false;

    @Override // annotated with @SignalMethod on the workflow interface
    public void subscriberAcknowledgedTerminator() {
      subscriberDone = true;
    }

    @Override
    public String complete(ChatInput input) {
      // ... do work and publish events ...

      // Returns true if the ack arrived, false on timeout. Either way, fall through.
      Workflow.await(Duration.ofSeconds(30), () -> subscriberDone);
      return result;
    }
  }
  ```

The full pattern is wired into the [Stream LLM output](#stream-llm-output) example below.

You can [inspect the terminal status](/workflow-streams#inspecting-terminal-status). A subscription ends cleanly when the Workflow reaches `COMPLETED`, `FAILED`, `CANCELED`, `TERMINATED`, or `TIMED_OUT`, but doesn't distinguish among them. If your application needs to know which (to display success or failure to the user, log the outcome, or decide whether to retry), call `workflowClient.newUntypedWorkflowStub(workflowId).describe()` after the subscription ends to inspect the Workflow's status.

## Stream from long-running Workflows 

Workflows that run for hours or accumulate thousands of events need to periodically roll over via [Continue-As-New](/workflow-streams#stream-from-long-running-workflows) to keep history bounded. Subscribers automatically follow these rollovers. To keep a stream running across them without subscribers seeing a gap, carry both your application state and the stream state across the boundary. Add a `WorkflowStreamState` field to your Workflow input, pass it to `WorkflowStream.newInstance`, and call `stream.continueAsNew(buildArgs)` to invoke the rollover. The helper drains waiting subscribers, waits for in-flight handlers to finish, snapshots the stream state, then continues-as-new with the arguments built by `buildArgs(postDrainState)`. It never returns:

```java
public class WorkflowInput {
  public int itemsProcessed;
  public WorkflowStreamState streamState;
}

public class LongRunningWorkflowImpl implements LongRunningWorkflow {
  private final WorkflowStream stream;
  private int itemsProcessed;

  @WorkflowInit
  public LongRunningWorkflowImpl(WorkflowInput input) {
    stream = WorkflowStream.newInstance(input.streamState);
    itemsProcessed = input.itemsProcessed;
  }

  @Override
  public void execute(WorkflowInput input) {
    while (true) {
      doOneIteration();
      itemsProcessed++;

      if (Workflow.getInfo().isContinueAsNewSuggested()) {
        stream.continueAsNew(
            state -> {
              WorkflowInput next = new WorkflowInput();
              next.itemsProcessed = itemsProcessed; // your own state, carried across
              next.streamState = state;             // the captured stream state
              return new Object[] {next};
            });
      }
    }
  }
}
```

The `streamState` field is `null` on a fresh start and a populated snapshot after a rollover. The `buildArgs` callback receives the post-detach `WorkflowStreamState` as its only argument, so the snapshot is guaranteed to happen *after* pollers detach.

To pass other Continue-As-New parameters such as a different Task Queue, or to use a custom publisher TTL, use the explicit recipe instead. Drain the pollers, wait for handlers to finish, snapshot the state with your chosen TTL, then call `Workflow.continueAsNew` yourself:

```java
import io.temporal.workflow.ContinueAsNewOptions;

stream.detachPollers();
Workflow.await(() -> Workflow.isEveryHandlerFinished());
WorkflowStreamState state = stream.getState(Duration.ofMinutes(30)); // custom publisher TTL
WorkflowInput next = new WorkflowInput();
next.itemsProcessed = itemsProcessed;
next.streamState = state;
Workflow.continueAsNew(
    ContinueAsNewOptions.newBuilder().setTaskQueue("other-tq").build(), next);
```

The carried `WorkflowStreamState` includes the entire in-memory log of the previous run, so streams that carry large items can hit Temporal's per-payload size limit at the rollover. Offload the bytes via [External Storage](/external-storage) so each item is a small reference rather than the full payload, and combine that with `stream.truncate(upToOffset)` to keep the carried log itself small.

## Deduplication window

See [How events are delivered](/workflow-streams#how-events-are-delivered) for more details on subscriber and publisher behavior. See [Tuning](/workflow-streams#tuning) for more details on how to change your settings to meet the requirements for your Workflow Streams.

There are two limits on the [deduplication window](/workflow-streams#deduplication-window) worth highlighting:

- **Publisher TTL.** At each Continue-As-New, deduplication entries whose last-seen time is older than this are dropped. The last-seen time is updated on each *successful* publish (not on each retry attempt), so a publisher that retries through a long partition without success can still age out. A publisher that returns after a longer pause may produce a duplicate. `stream.continueAsNew(...)` snapshots with a 15-minute default; to tune it, use the explicit recipe above and pass your value to `getState(publisherTtl)`.
- **`maxRetryDuration`.** A `WorkflowStreamClient` retries a failed batch for up to this long (default 10 minutes). If the duration elapses with the batch still pending, the client gives up, the pending batch is dropped, and a `FlushTimeoutException` is raised.

  ```java
  WorkflowStreamClientOptions.newBuilder()
      .setMaxRetryDuration(Duration.ofMinutes(10))
      .build();
  ```

  On timeout, the dropped batch is at-most-once: it may or may not have reached the Workflow. One operational caveat: the `FlushTimeoutException` is raised from inside the background flusher and terminates it. Until you call `client.flush()` or `client.close()` — which surface the deferred exception — subsequent publishes accumulate in the buffer with no flusher to ship them. `maxRetryDuration` must be less than the Workflow's publisher TTL to preserve exactly-once delivery.

## Best practices

There are a few details to note if you're writing custom message handlers or testing the library's capabilities:

- **Construct exactly one `WorkflowStream` per Workflow, preferably in `@WorkflowInit`.** The factory registers the publish Signal, poll Update, and offset Query handlers on the current Workflow. A `@WorkflowInit` constructor runs before any handler dispatch, so polls and offset Queries arriving with the first Workflow Task are accepted; construction at the top of the Workflow method leaves them rejected until the stream exists.
- **`item.getPayload()` is always the raw payload.** Decode it with a converter built from the same payload converters used by the publisher. When publishers and subscribers both rely on the defaults, `DefaultDataConverter.STANDARD_INSTANCE` matches on both sides. If you pass `setPayloadConverters` on the Workflow side or the client side, build a matching converter on the subscriber side.
- **The codec chain runs once on the envelope.** Payload codecs (encryption, compression) configured on the Temporal client run on the Signal or Update envelope that carries each batch, never per item, so items are never double-encoded. `setPayloadConverters` handles only per-item serialization; its `PayloadConverter[]` type makes it impossible to slot a codec in per item.
- **Listener callbacks must not block.** They run serialized on the client's poll executor, which drives every subscription on the client. Blocking in `onNext` stalls the client's other subscriptions. Hand slow work to your own executor and return the resulting `CompletionStage` for backpressure.
- **Size the poll executor for slow Workflows.** The default executor has 2 daemon threads, is created lazily, and is owned (and shut down) by the client; a user-supplied executor is never shut down by the client. The executor runs the short update-admission and delivery steps and poll cooldowns, never the long poll itself, so a small pool serves many subscriptions. The known worst case for pool pressure is a backlogged Workflow pinning a thread in the update-admission call; supply a bigger pool via `setPollExecutor` when running many subscriptions against slow Workflows.
- **Cross-language interop depends on the configured data converter.** The handler names, JSON envelope field names, and per-item payload encoding match the other SDKs' packages exactly, so a Java publisher or subscriber interoperates with a Workflow written in any of them and vice versa. One Java-specific caveat: the protocol envelope types are serialized by the Workflow's and client's *configured* data converter. The default Jackson JSON converter produces the wire-compatible snake_case field names; if you configure a non-Jackson JSON converter, it must produce the same field names for cross-language interop.

## Example: Stream LLM output 

The headline use case fits the publish/subscribe shapes documented above. An Activity calls the model and publishes deltas as they arrive. The Workflow starts the Activity and waits for the consumer to acknowledge end-of-stream. The consumer subscribes, accumulates the deltas, and clears its accumulated state on `RETRY` before continuing. The shape works for a terminal client, a desktop UI, or a Server-Sent Events (SSE) endpoint forwarding to a browser. Anything that holds the displayed state calls `render()` to display it.

If your Activity can retry, the consumer side has to account for it. A retried attempt is a fresh publisher, so its output appears in the stream alongside the output from the previous attempt. In the LLM streaming pattern below, that means the failed attempt's partial deltas and the retried attempt's full output both reach a subscribed UI unless the UI resets on a `RETRY` event. The example wires up that pattern. See [How events are delivered](/workflow-streams#how-events-are-delivered) for the precise guarantees.

**LlmActivitiesImpl.java**

```java
import com.openai.client.OpenAIClient;
import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.core.http.StreamResponse;
import com.openai.models.chat.completions.ChatCompletionChunk;
import com.openai.models.chat.completions.ChatCompletionCreateParams;
import io.temporal.activity.Activity;
import io.temporal.workflowstreams.TopicHandle;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamClientOptions;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;

public class TextDelta {
  public String text;

  public TextDelta() {}

  public TextDelta(String text) {
    this.text = text;
  }
}

public class RetryEvent {
  public int attempt;

  public RetryEvent() {}

  public RetryEvent(int attempt) {
    this.attempt = attempt;
  }
}

public class CloseEvent {}

public class LlmActivitiesImpl implements LlmActivities {
  @Override
  public String streamCompletion(String prompt) {
    WorkflowStreamClientOptions options =
        WorkflowStreamClientOptions.newBuilder()
            .setBatchInterval(Duration.ofMillis(200))
            .build();
    try (WorkflowStreamClient streamClient = WorkflowStreamClient.fromActivity(options)) {
      TopicHandle deltas = streamClient.topic("delta");
      TopicHandle retry = streamClient.topic("retry");
      TopicHandle close = streamClient.topic("close");

      // Tell consumers an earlier attempt's deltas are stale.
      int attempt = Activity.getExecutionContext().getInfo().getAttempt();
      if (attempt > 1) {
        retry.publish(new RetryEvent(attempt), /* forceFlush */ true);
      }

      // Disable provider-side retries; let Temporal own retry policy at the Activity layer.
      OpenAIClient openai = OpenAIOkHttpClient.builder().fromEnv().maxRetries(0).build();
      ChatCompletionCreateParams params =
          ChatCompletionCreateParams.builder().model("gpt-4o-mini").addUserMessage(prompt).build();

      StringBuilder full = new StringBuilder();
      AtomicBoolean first = new AtomicBoolean(true);
      try (StreamResponse<ChatCompletionChunk> stream =
          openai.chat().completions().createStreaming(params)) {
        stream.stream()
            .forEach(
                chunk ->
                    chunk.choices().stream()
                        .findFirst()
                        .flatMap(choice -> choice.delta().content())
                        .filter(text -> !text.isEmpty())
                        .ifPresent(
                            text -> {
                              // forceFlush only on the first delta so the user sees something
                              // immediately; subsequent deltas batch at the 200 ms interval.
                              deltas.publish(new TextDelta(text), first.getAndSet(false));
                              full.append(text);
                            }));
      }
      close.publish(new CloseEvent());
      return full.toString();
    }
  }
}
```

**ChatWorkflowImpl.java**

```java
import io.temporal.activity.ActivityOptions;
import io.temporal.workflow.SignalMethod;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInit;
import io.temporal.workflow.WorkflowInterface;
import io.temporal.workflow.WorkflowMethod;
import io.temporal.workflowstreams.WorkflowStream;
import java.time.Duration;

@WorkflowInterface
public interface ChatWorkflow {
  @WorkflowMethod
  String complete(ChatInput input);

  @SignalMethod
  void subscriberAcknowledgedTerminator();
}

public class ChatWorkflowImpl implements ChatWorkflow {
  private final LlmActivities activities =
      Workflow.newActivityStub(
          LlmActivities.class,
          ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofMinutes(5)).build());

  private boolean subscriberDone = false;

  @WorkflowInit
  public ChatWorkflowImpl(ChatInput input) {
    WorkflowStream.newInstance(input.streamState);
  }

  @Override
  public void subscriberAcknowledgedTerminator() {
    subscriberDone = true;
  }

  @Override
  public String complete(ChatInput input) {
    String result = activities.streamCompletion(input.prompt);

    // Wait for the subscriber to ack the terminal `close` event. The timeout
    // is a fallback for when no subscriber is attached; with the ack, the
    // typical case exits as soon as the subscriber confirms.
    Workflow.await(Duration.ofSeconds(30), () -> subscriberDone);
    return result;
  }
}
```

**Consumer.java**

```java
import io.temporal.client.WorkflowClient;
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.workflowstreams.SubscribeOptions;
import io.temporal.workflowstreams.WorkflowStreamClient;
import io.temporal.workflowstreams.WorkflowStreamItem;
import io.temporal.workflowstreams.WorkflowStreamSubscription;

public String streamChat(WorkflowClient workflowClient, String chatId) {
  StringBuilder output = new StringBuilder();

  Runnable render =
      () -> {
        // ... display the accumulated output (terminal redraw, UI update, etc.)
      };

  SubscribeOptions options =
      SubscribeOptions.newBuilder().setTopics("delta", "retry", "close").build();
  try (WorkflowStreamClient stream = WorkflowStreamClient.newInstance(workflowClient, chatId);
      WorkflowStreamSubscription subscription = stream.subscribe(options)) {
    for (WorkflowStreamItem item : subscription) {
      switch (item.getTopic()) {
        case "retry":
          // Earlier attempt's deltas are stale; drop what we've shown.
          output.setLength(0);
          render.run();
          break;
        case "delta":
          TextDelta delta =
              DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
                  item.getPayload(), TextDelta.class, TextDelta.class);
          output.append(delta.text);
          render.run();
          break;
        case "close":
          // Acknowledge so the Workflow can return without waiting on the fallback timeout.
          workflowClient
              .newUntypedWorkflowStub(chatId)
              .signal("subscriberAcknowledgedTerminator");
          return output.toString();
      }
    }
  }

  return output.toString();
}
```

A few choices in this shape are deliberate:

- The Activity is the publisher because it owns the non-deterministic LLM call. The Workflow processes only the Activity's return value, never reading its own stream. See [Publish from a client](#publish-from-a-client) for why.
- The Activity publishes a `RETRY` event when `Activity.getExecutionContext().getInfo().getAttempt() > 1`. This lets the UI respond appropriately to the failure, typically by clearing accumulated deltas before the next attempt's deltas arrive (see [How events are delivered](/workflow-streams#how-events-are-delivered)).
- Termination uses an *ack handshake*: the consumer signals the Workflow once it has received the `close` event, so the Workflow can return as soon as the subscriber confirms. The `Workflow.await` timeout is the fallback when no subscriber is attached (see [Closing the stream](#closing-the-stream) for the simpler fixed-sleep alternative).
- `forceFlush` is `true` only on the first delta and on the `RETRY` sentinel, where latency matters. Subsequent deltas batch at the 200 ms `batchInterval`; per-delta `forceFlush` would generate one Signal per token (see [Tuning](/workflow-streams#tuning) for the trade-off).

## See also

- [Workflow Streams samples (samples-java)](https://github.com/temporalio/samples-java/tree/main/core/src/main/java/io/temporal/samples/workflowstreams): six runnable scenarios covering basic publish/subscribe, the non-blocking listener, reconnecting subscribers, external publishers, bounded logs, and LLM streaming.
- [`temporal-workflowstreams` API reference](https://javadoc.io/doc/io.temporal/temporal-workflowstreams).
- [Workflow message passing](/develop/java/workflows/message-passing): Signals, Updates, and Queries that Workflow Streams is built on.
- [Converters and encryption](/develop/java/best-practices/converters-and-encryption): converters and codecs.