Add transfer type converters

contrib/temporal-workflowstreams/README.md

@@ -124,9 +124,13 @@ automatically follows continue-as-new chains, recovers from truncation by
 restarting from the current base offset, and also ends when the owning
 `WorkflowStreamClient` is closed.
 
-Items carry the raw `io.temporal.api.common.v1.Payload`; decode at the call
-site with your data converter. Offsets are **global** (across all topics), not
-per-topic.
+Items carry the raw `io.temporal.api.common.v1.Payload`. Use
+`WorkflowStreamClient.decodeItem` to decode them with the configured stream
+item converter. Transfer conversion and payload conversion apply to each item.
+Payload codecs apply only once to the surrounding Temporal signal or update
+envelope, never to an individual item. Configure matching payload converters
+on the workflow and client sides. Offsets are **global** (across all topics),
+not per-topic.
 
 ### Listener (non-blocking)
 
@@ -147,9 +151,7 @@ WorkflowStreamSubscriptionHandle handle =
         new WorkflowStreamListener() {
           @Override
           public CompletionStage<Void> onNext(WorkflowStreamItem item) {
-            String value =
-                DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
-                    item.getPayload(), String.class, String.class);
+            String value = client.decodeItem(item, String.class);
             System.out.printf(
                 "offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value);
             return null; // or a pending stage to apply backpressure
@@ -176,9 +178,7 @@ polling still runs on the shared executor:
 ```java
 try (WorkflowStreamSubscription subscription = client.subscribe(options)) {
   for (WorkflowStreamItem item : subscription) {
-    String value =
-        DefaultDataConverter.STANDARD_INSTANCE.fromPayload(
-            item.getPayload(), String.class, String.class);
+    String value = client.decodeItem(item, String.class);
     System.out.printf("offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value);
   }
 }
@@ -194,7 +194,7 @@ unrecoverable poll failure is rethrown from `hasNext()`.
 | `batchInterval` | 2s | Automatic flush interval |
 | `maxBatchSize` | unset | Flush once the buffer reaches this size |
 | `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 |
+| `payloadConverters` | standard set | Per-item transfer and payload conversion. 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 |
 | `SubscribeOptions.pollCooldown` | 100ms | Min interval between polls |
 

contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java

@@ -3,7 +3,6 @@
 import io.temporal.api.common.v1.Payload;
 import io.temporal.common.Experimental;
 import io.temporal.common.converter.DataConverter;
-import io.temporal.common.converter.DefaultDataConverter;
 import io.temporal.failure.ApplicationFailure;
 import io.temporal.workflow.ContinueAsNewOptions;
 import io.temporal.workflow.Workflow;
@@ -77,13 +76,7 @@ public static WorkflowStream newInstance(
   }
 
   private WorkflowStream(@Nullable WorkflowStreamState priorState, WorkflowStreamOptions options) {
-    // A converter built only from PayloadConverters is codec-free, so workflow-published
-    // items are never double-encoded against the worker's response codec.
-    if (options.getPayloadConverters().length > 0) {
-      this.dataConverter = new DefaultDataConverter(options.getPayloadConverters());
-    } else {
-      this.dataConverter = DefaultDataConverter.STANDARD_INSTANCE;
-    }
+    this.dataConverter = WorkflowStreamDataConverter.create(options.getPayloadConverters());
 
     if (priorState != null) {
       baseOffset = priorState.baseOffset;

contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java

@@ -6,9 +6,9 @@
 import io.temporal.client.WorkflowStub;
 import io.temporal.common.Experimental;
 import io.temporal.common.converter.DataConverter;
-import io.temporal.common.converter.DefaultDataConverter;
 import io.temporal.workflowstreams.internal.StreamPublisher;
 import io.temporal.workflowstreams.internal.SubscriptionDriver;
+import java.lang.reflect.Type;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -32,6 +32,7 @@ public final class WorkflowStreamClient implements AutoCloseable {
   private final WorkflowClient client;
   private final String workflowId;
   private final StreamPublisher publisher;
+  private final DataConverter itemDataConverter;
   @Nullable private final ScheduledExecutorService userPollExecutor;
 
   private final Map<String, TopicHandle> topicHandles = new HashMap<>();
@@ -83,22 +84,15 @@ private WorkflowStreamClient(
     this.client = client;
     this.workflowId = workflowId;
 
-    // A converter built only from PayloadConverters is codec-free, so items are never
-    // double-encoded against the codec on the client's signal/update envelope.
-    DataConverter dataConverter;
-    if (options.getPayloadConverters().length > 0) {
-      dataConverter = new DefaultDataConverter(options.getPayloadConverters());
-    } else {
-      dataConverter = DefaultDataConverter.STANDARD_INSTANCE;
-    }
+    this.itemDataConverter = WorkflowStreamDataConverter.create(options.getPayloadConverters());
 
     this.userPollExecutor = options.getPollExecutor();
 
     WorkflowStub stub = client.newUntypedWorkflowStub(workflowId);
     this.publisher =
         new StreamPublisher(
             input -> stub.signal(WorkflowStreamConstants.PUBLISH_SIGNAL_NAME, input),
-            dataConverter,
+            itemDataConverter,
             options.getBatchInterval(),
             options.getMaxBatchSize(),
             options.getMaxRetryDuration());
@@ -112,6 +106,16 @@ public synchronized TopicHandle topic(String name) {
     return topicHandles.computeIfAbsent(name, n -> new TopicHandle(n, this));
   }
 
+  /** Decodes an item using this client's configured, codec-free item converter. */
+  public <T> T decodeItem(WorkflowStreamItem item, Class<T> valueClass) {
+    return decodeItem(item, valueClass, valueClass);
+  }
+
+  /** Decodes an item using this client's configured, codec-free item converter. */
+  public <T> T decodeItem(WorkflowStreamItem item, Class<T> valueClass, Type valueType) {
+    return itemDataConverter.fromPayload(item.getPayload(), valueClass, valueType);
+  }
+
   /**
    * Sends buffered (and pending) items and waits for server confirmation. Returns once the items
    * buffered at call time have been signaled to the workflow and acknowledged.
@@ -141,9 +145,9 @@ public long getOffset() {
    * }</pre>
    *
    * <p>The consuming thread blocks waiting for items; polling itself runs on the client's poll
-   * executor. Each item carries the raw {@link io.temporal.api.common.v1.Payload}; decode it with
-   * your data converter. The subscription ends cleanly when the workflow reaches a terminal state,
-   * automatically follows continue-as-new chains, and also ends when this client is closed.
+   * executor. Decode each item with {@link #decodeItem(WorkflowStreamItem, Class)}. The
+   * subscription ends cleanly when the workflow reaches a terminal state, automatically follows
+   * continue-as-new chains, and also ends when this client is closed.
    */
   public WorkflowStreamSubscription subscribe(SubscribeOptions options) {
     return new WorkflowStreamSubscription(listener -> newSubscriptionDriver(options, listener));

contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java

@@ -101,8 +101,9 @@ public Builder setMaxRetryDuration(Duration maxRetryDuration) {
      * <p>Only payload conversion happens here — never a payload codec (encryption, compression).
      * The codec chain configured on the Temporal client runs once on the signal/update envelope
      * that carries each batch, so encoding items here too would double-encode them; the {@code
-     * PayloadConverter[]} type makes that mistake impossible. To decode subscribed items, use a
-     * converter built from the same payload converters.
+     * PayloadConverter[]} type makes that mistake impossible. Transfer conversion and payload
+     * conversion apply to each item. Decode subscribed items with {@link
+     * WorkflowStreamClient#decodeItem(WorkflowStreamItem, Class)}.
      *
      * <p>Default: the standard converter set.
      */

contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamDataConverter.java

@@ -0,0 +1,27 @@
+package io.temporal.workflowstreams;
+
+import io.temporal.common.converter.DataConverter;
+import io.temporal.common.converter.DefaultDataConverter;
+import io.temporal.common.converter.PayloadConverter;
+import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter;
+
+/**
+ * Builds the converter used for individual Workflow Stream items.
+ *
+ * <p>Items are serialized into the stream protocol envelope, which is then sent through a Temporal
+ * signal or update. That outer envelope already uses the workflow client's or worker's configured
+ * data converter and payload codecs. Applying codecs to an item here would encode it a second time
+ * and prevent subscribers from decoding the raw item payload. This converter therefore includes
+ * only the configured payload converters, while still applying SDK-managed transfer conversion.
+ */
+final class WorkflowStreamDataConverter {
+  private WorkflowStreamDataConverter() {}
+
+  static DataConverter create(PayloadConverter[] payloadConverters) {
+    DataConverter converter =
+        payloadConverters.length == 0
+            ? DefaultDataConverter.STANDARD_INSTANCE
+            : new DefaultDataConverter(payloadConverters);
+    return TemporalTransferTypeDataConverter.wrap(converter);
+  }
+}

contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java

@@ -4,10 +4,9 @@
 import io.temporal.common.Experimental;
 
 /**
- * A single decoded item yielded by a subscription. {@code payload} is the raw {@link Payload};
- * decode it at the call site with a payload converter, e.g. {@code
- * DefaultDataConverter.STANDARD_INSTANCE.fromPayload(item.getPayload(), String.class,
- * String.class)}.
+ * A single decoded item yielded by a subscription. {@code payload} is the raw {@link Payload}; use
+ * {@link WorkflowStreamClient#decodeItem(WorkflowStreamItem, Class)} to decode it with the stream
+ * client's configured item converter.
  */
 @Experimental
 public final class WorkflowStreamItem {

contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamOptions.java

@@ -40,7 +40,8 @@ private Builder() {}
      * <p>As on the client side, only payload conversion happens here — never a payload codec. The
      * worker's codec chain runs once on the poll-update response that carries each batch to
      * subscribers, so encoding items here too would double-encode them; the {@code
-     * PayloadConverter[]} type makes that impossible.
+     * PayloadConverter[]} type makes that impossible. Transfer conversion and payload conversion
+     * apply to each item.
      *
      * <p>There is no public accessor for the worker's configured data converter inside workflow
      * code, so it cannot be picked up automatically; pass the matching payload converters here to

contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java

@@ -5,6 +5,7 @@
 import io.temporal.client.WorkflowStub;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.DefaultDataConverter;
+import io.temporal.common.converter.EncodingKeys;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflow;
 import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflowImpl;
@@ -69,6 +70,48 @@ public void testSubscribeDeliversItemsAndAdvancesOffset() {
     stub.getResult(Void.class);
   }
 
+  @Test
+  public void testClientPublishedItemUsesTransferConverter() {
+    WorkflowStub stub = startHostWorkflow();
+    try (WorkflowStreamClient streamClient = newStreamClient(stub)) {
+      streamClient.topic("evt").publish(new TransferStreamTestModel("client"), true);
+      streamClient.flush();
+
+      try (WorkflowStreamSubscription subscription = streamClient.subscribe(FAST_POLL)) {
+        WorkflowStreamItem item = subscription.next();
+        Assert.assertEquals(
+            "The configured payload converter must receive the protobuf transfer representation",
+            "json/protobuf",
+            item.getPayload()
+                .getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY)
+                .toStringUtf8());
+        TransferStreamTestModel result =
+            streamClient.decodeItem(item, TransferStreamTestModel.class);
+        Assert.assertEquals(new TransferStreamTestModel("client"), result);
+        Assert.assertTrue(result.wasTransferred());
+      }
+    }
+    stub.signal("finish");
+    stub.getResult(Void.class);
+  }
+
+  @Test
+  public void testWorkflowPublishedItemUsesTransferConverter() {
+    WorkflowStub stub = startHostWorkflow();
+    try (WorkflowStreamClient streamClient = newStreamClient(stub)) {
+      stub.signal("publishTransfer", "evt", new TransferStreamTestModel("workflow"));
+
+      try (WorkflowStreamSubscription subscription = streamClient.subscribe(FAST_POLL)) {
+        TransferStreamTestModel result =
+            streamClient.decodeItem(subscription.next(), TransferStreamTestModel.class);
+        Assert.assertEquals(new TransferStreamTestModel("workflow"), result);
+        Assert.assertTrue(result.wasTransferred());
+      }
+    }
+    stub.signal("finish");
+    stub.getResult(Void.class);
+  }
+
   @Test
   public void testTopicHandleSubscribeFilters() {
     WorkflowStub stub = startHostWorkflow();

contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java

@@ -25,6 +25,9 @@ public interface SubscribeHostWorkflow {
     @SignalMethod
     void publishLocal(String topic, String value);
 
+    @SignalMethod
+    void publishTransfer(String topic, TransferStreamTestModel value);
+
     @UpdateMethod
     void truncate(long upToOffset);
   }
@@ -66,6 +69,11 @@ public void publishLocal(String topic, String value) {
       stream.topic(topic).publish(value);
     }
 
+    @Override
+    public void publishTransfer(String topic, TransferStreamTestModel value) {
+      stream.topic(topic).publish(value);
+    }
+
     @Override
     public void truncate(long upToOffset) {
       stream.truncate(upToOffset);

contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/TransferStreamTestModel.java

@@ -0,0 +1,56 @@
+package io.temporal.workflowstreams;
+
+import com.google.protobuf.StringValue;
+import io.temporal.common.converter.TransferTypeConverter;
+import io.temporal.common.converter.TransferTypeConvertible;
+import java.lang.reflect.Type;
+
+/** Test model that proves a Workflow Stream item uses transfer conversion. */
+@TransferTypeConvertible(TransferStreamTestModel.Converter.class)
+public final class TransferStreamTestModel {
+  private final String value;
+  private final boolean transferred;
+
+  public TransferStreamTestModel(String value) {
+    this(value, false);
+  }
+
+  private TransferStreamTestModel(String value, boolean transferred) {
+    this.value = value;
+    this.transferred = transferred;
+  }
+
+  public boolean wasTransferred() {
+    return transferred;
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    return other instanceof TransferStreamTestModel
+        && value.equals(((TransferStreamTestModel) other).value);
+  }
+
+  @Override
+  public int hashCode() {
+    return value.hashCode();
+  }
+
+  public static final class Converter implements TransferTypeConverter<TransferStreamTestModel> {
+    public Converter() {}
+
+    @Override
+    public Type getTransferType(Type valueType) {
+      return StringValue.class;
+    }
+
+    @Override
+    public Object toTransferType(TransferStreamTestModel value) {
+      return StringValue.of(value.value);
+    }
+
+    @Override
+    public TransferStreamTestModel fromTransferType(Object value, Type valueType) {
+      return new TransferStreamTestModel(((StringValue) value).getValue(), true);
+    }
+  }
+}