Skip to main content

Workflow Streams - Java SDK

View Markdown

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 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.

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).

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 with stream.topic(name), then call publish() on the returned WorkflowTopicHandle to append events. Repeated calls with the same name return the same handle.

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). 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:

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.

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:

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 (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:

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 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:

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 pull from the Workflow's log on their own schedule, so a slow subscriber doesn't slow down publishers. 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 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.

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.

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:

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 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. Sleep between the terminator and the return so any in-flight poll has time to fetch the terminator before the Workflow exits:

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

  • 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:

    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 example below.

You can inspect the 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 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:

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:

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 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 for more details on subscriber and publisher behavior. See 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 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.

    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 for the precise guarantees.

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();
    }
  }
}

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 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).
  • 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 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 for the trade-off).

See also​