Add transfer type converters

CHANGELOG.md

@@ -25,6 +25,11 @@ to docs, or any other relevant information.
   `WorkflowOutboundCallsInterceptorBase` need to add them, and Base subclasses override them to see calls with options.
 
 ### Added
+- Applications can use the experimental `@TransferTypeConvertible` annotation and
+  `TransferTypeConverter` interface to convert top-level model values to transfer types. The SDK
+  applies this conversion to workflow, activity, Nexus, and schedule payloads, including failure
+  details. `WorkflowClient.getOptions().getDataConverter()` returns the transfer-aware converter
+  that the client uses.
 - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and
   `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time
   child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure`

temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java

@@ -3,11 +3,13 @@
 import com.uber.m3.tally.Scope;
 import io.temporal.api.common.v1.Payload;
 import io.temporal.common.context.ContextPropagator;
+import io.temporal.common.converter.DataConverter;
 import io.temporal.common.interceptors.ActivityClientCallsInterceptor;
 import io.temporal.common.interceptors.ActivityClientInterceptor;
 import io.temporal.common.interceptors.Header;
 import io.temporal.internal.client.ActivityClientInternal;
 import io.temporal.internal.client.ActivityHandleImpl;
+import io.temporal.internal.client.ClientDataConverterFactory;
 import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
 import io.temporal.internal.client.RootActivityClientInvoker;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
@@ -38,6 +40,8 @@ class ActivityClientImpl implements ActivityClient, ActivityClientInternal {
   private final Scope metricsScope;
 
   ActivityClientImpl(WorkflowServiceStubs stubs, ActivityClientOptions options) {
+    DataConverter internalDataConverter =
+        ClientDataConverterFactory.forClient(options.getDataConverter());
     this.stubs = stubs;
     this.options = options;
     WorkflowServiceStubs namespacedStubs =
@@ -46,18 +50,19 @@ class ActivityClientImpl implements ActivityClient, ActivityClientInternal {
         stubs.getOptions().getMetricsScope().tagged(MetricsTag.defaultTags(options.getNamespace()));
     GenericWorkflowClientImpl genericClient =
         new GenericWorkflowClientImpl(namespacedStubs, metricsScope);
-    this.invoker = initializeClientInvoker(genericClient, options);
+    this.invoker = initializeClientInvoker(genericClient, options, internalDataConverter);
     this.manualActivityCompletionClientFactory =
         ManualActivityCompletionClientFactory.newFactory(
-            namespacedStubs,
-            options.getNamespace(),
-            options.getIdentity(),
-            options.getDataConverter());
+            namespacedStubs, options.getNamespace(), options.getIdentity(), internalDataConverter);
   }
 
   private static ActivityClientCallsInterceptor initializeClientInvoker(
-      GenericWorkflowClientImpl genericClient, ActivityClientOptions options) {
-    ActivityClientCallsInterceptor invoker = new RootActivityClientInvoker(genericClient, options);
+      GenericWorkflowClientImpl genericClient,
+      ActivityClientOptions options,
+      DataConverter dataConverter) {
+    ActivityClientCallsInterceptor invoker =
+        new RootActivityClientInvoker(
+            genericClient, options.getNamespace(), options.getIdentity(), dataConverter);
     for (ActivityClientInterceptor interceptor : options.getInterceptors()) {
       invoker = interceptor.activityClientCallsInterceptor(invoker);
     }

temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java

@@ -129,7 +129,7 @@ public <V> void reportCancellationStandalone(
 
   @Override
   public <V> void heartbeat(byte[] taskToken, V details) throws ActivityCompletionException {
-    factory.getClient(taskToken, metricsScope).recordHeartbeat(details);
+    factory.getClient(taskToken, metricsScope, serializationContext).recordHeartbeat(details);
   }
 
   @Override

temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java

@@ -11,12 +11,14 @@
 import io.temporal.common.interceptors.NexusClientCallsInterceptor.ListNexusOperationExecutionsOutput;
 import io.temporal.common.interceptors.NexusClientInterceptor;
 import io.temporal.internal.WorkflowThreadMarker;
+import io.temporal.internal.client.ClientDataConverterFactory;
 import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
 import io.temporal.internal.client.NexusClientResolvedOptions;
 import io.temporal.internal.client.NexusOperationHandleImpl;
 import io.temporal.internal.client.RootNexusClientInvoker;
 import io.temporal.internal.client.external.GenericWorkflowClient;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
+import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import java.util.List;
@@ -40,10 +42,17 @@ public class NexusClientImpl implements NexusClient {
   public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientOptions options) {
     enforceNonWorkflowThread();
     return WorkflowThreadMarker.protectFromWorkflowThread(
-        new NexusClientImpl(service, options.toResolvedOptions()), NexusClient.class);
+        new NexusClientImpl(service, options), NexusClient.class);
   }
 
-  NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientResolvedOptions options) {
+  NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientOptions clientOptions) {
+    NexusClientResolvedOptions options =
+        clientOptions.toResolvedOptions(
+            clientOptions.getExternalStorage() == null
+                ? ClientDataConverterFactory.forClient(clientOptions.getDataConverter())
+                : ClientDataConverterFactory.forClient(
+                    clientOptions.getDataConverter(),
+                    ExternalStorageRunner.create(clientOptions.getExternalStorage())));
     workflowServiceStubs =
         new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace());
     this.workflowServiceStubs = workflowServiceStubs;

temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java

@@ -5,8 +5,6 @@
 import io.temporal.common.converter.GlobalDataConverter;
 import io.temporal.common.interceptors.NexusClientInterceptor;
 import io.temporal.internal.client.NexusClientResolvedOptions;
-import io.temporal.internal.payload.storage.ExternalStorageDataConverter;
-import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import io.temporal.payload.storage.ExternalStorage;
 import java.lang.management.ManagementFactory;
 import java.util.Collections;
@@ -92,16 +90,11 @@ public String getIdentity() {
    * Converts this {@link NexusClientOptions} instance into a {@link NexusClientResolvedOptions}
    * instance, which contains the fully resolved runtime settings used by the internal Nexus client.
    *
+   * @param internalDataConverter converter assembled for internal client use
    * @return a {@link NexusClientResolvedOptions} instance with the resolved options
    */
-  NexusClientResolvedOptions toResolvedOptions() {
-    DataConverter resolvedDataConverter = dataConverter;
-    if (externalStorage != null) {
-      resolvedDataConverter =
-          new ExternalStorageDataConverter(
-              resolvedDataConverter, ExternalStorageRunner.create(externalStorage));
-    }
-    return new NexusClientResolvedOptions(namespace, interceptors, resolvedDataConverter, identity);
+  NexusClientResolvedOptions toResolvedOptions(DataConverter internalDataConverter) {
+    return new NexusClientResolvedOptions(namespace, interceptors, internalDataConverter, identity);
   }
 
   /** Returns a fresh builder. */

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java

@@ -14,10 +14,12 @@
 import io.temporal.api.workflowservice.v1.*;
 import io.temporal.client.WorkflowInvocationHandler.InvocationType;
 import io.temporal.common.WorkflowExecutionHistory;
+import io.temporal.common.converter.DataConverter;
 import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
 import io.temporal.common.interceptors.WorkflowClientInterceptor;
 import io.temporal.internal.WorkflowThreadMarker;
 import io.temporal.internal.client.*;
+import io.temporal.internal.client.ClientDataConverterFactory;
 import io.temporal.internal.client.NexusStartWorkflowResponse;
 import io.temporal.internal.client.external.GenericWorkflowClient;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
@@ -52,6 +54,7 @@ final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClient
 
   private final GenericWorkflowClient genericClient;
   private final WorkflowClientOptions options;
+  private final DataConverter internalDataConverter;
   private final ManualActivityCompletionClientFactory manualActivityCompletionClientFactory;
   private final WorkflowClientCallsInterceptor workflowClientCallsInvoker;
   private final WorkflowServiceStubs workflowServiceStubs;
@@ -103,9 +106,10 @@ public static WorkflowClient newInstance(
     // Set merged plugins after configuration, then validate
     builder.setPlugins(mergedPlugins);
     options = builder.validateAndBuildWithDefaults();
+    this.internalDataConverter = ClientDataConverterFactory.forClient(options.getDataConverter());
+    this.options = options;
     workflowServiceStubs =
         new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace());
-    this.options = options;
     this.workflowServiceStubs = workflowServiceStubs;
     this.metricsScope =
         workflowServiceStubs
@@ -124,7 +128,7 @@ public static WorkflowClient newInstance(
             workflowServiceStubs,
             options.getNamespace(),
             options.getIdentity(),
-            options.getDataConverter(),
+            getInternalDataConverter(),
             externalStorageRunner);
 
     java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval();
@@ -142,7 +146,14 @@ public static WorkflowClient newInstance(
   private WorkflowClientCallsInterceptor initializeClientInvoker() {
     WorkflowClientCallsInterceptor workflowClientInvoker =
         new RootWorkflowClientInvoker(
-            genericClient, options, workerFactoryRegistry, externalStorageRunner);
+            genericClient,
+            options.getNamespace(),
+            options.getIdentity(),
+            options.getQueryRejectCondition(),
+            options.getContextPropagators(),
+            internalDataConverter,
+            workerFactoryRegistry,
+            externalStorageRunner);
     for (WorkflowClientInterceptor clientInterceptor : interceptors) {
       workflowClientInvoker =
           clientInterceptor.workflowClientCallsInterceptor(workflowClientInvoker);
@@ -160,13 +171,18 @@ public WorkflowClientOptions getOptions() {
     return options;
   }
 
+  @Override
+  public DataConverter getInternalDataConverter() {
+    return internalDataConverter;
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   public <T> T newWorkflowStub(Class<T> workflowInterface, WorkflowOptions options) {
     checkAnnotation(workflowInterface, WorkflowMethod.class);
     WorkflowInvocationHandler invocationHandler =
         new WorkflowInvocationHandler(
-            workflowInterface, this.getOptions(), workflowClientCallsInvoker, options);
+            workflowInterface, this.options, workflowClientCallsInvoker, options);
     return (T)
         Proxy.newProxyInstance(
             workflowInterface.getClassLoader(),
@@ -243,7 +259,7 @@ public <T> T newWorkflowStub(
     WorkflowInvocationHandler invocationHandler =
         new WorkflowInvocationHandler(
             workflowInterface,
-            this.getOptions(),
+            options,
             workflowClientCallsInvoker,
             execution.build(),
             legacyTargeting,

temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java

@@ -3,9 +3,11 @@
 import static io.temporal.internal.WorkflowThreadMarker.enforceNonWorkflowThread;
 
 import com.uber.m3.tally.Scope;
+import io.temporal.common.converter.DataConverter;
 import io.temporal.common.interceptors.ScheduleClientCallsInterceptor;
 import io.temporal.common.interceptors.ScheduleClientInterceptor;
 import io.temporal.internal.WorkflowThreadMarker;
+import io.temporal.internal.client.ClientDataConverterFactory;
 import io.temporal.internal.client.NamespaceInjectWorkflowServiceStubs;
 import io.temporal.internal.client.RootScheduleClientInvoker;
 import io.temporal.internal.client.external.GenericWorkflowClient;
@@ -71,6 +73,8 @@ public static ScheduleClient newInstance(
     // Set merged plugins after configuration, then build
     builder.setPlugins(mergedPlugins);
     options = builder.build();
+    DataConverter internalDataConverter =
+        ClientDataConverterFactory.forClient(options.getDataConverter());
 
     workflowServiceStubs =
         new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace());
@@ -83,7 +87,7 @@ public static ScheduleClient newInstance(
             .tagged(MetricsTag.defaultTags(options.getNamespace()));
     this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
     this.interceptors = options.getInterceptors();
-    this.scheduleClientCallsInvoker = initializeClientInvoker();
+    this.scheduleClientCallsInvoker = initializeClientInvoker(internalDataConverter);
   }
 
   private static ScheduleClientPlugin[] extractScheduleClientPlugins(
@@ -100,9 +104,14 @@ private static ScheduleClientPlugin[] extractScheduleClientPlugins(
     return schedulePlugins.toArray(new ScheduleClientPlugin[0]);
   }
 
-  private ScheduleClientCallsInterceptor initializeClientInvoker() {
+  private ScheduleClientCallsInterceptor initializeClientInvoker(DataConverter dataConverter) {
     ScheduleClientCallsInterceptor scheduleClientInvoker =
-        new RootScheduleClientInvoker(genericClient, options);
+        new RootScheduleClientInvoker(
+            genericClient,
+            options.getNamespace(),
+            options.getIdentity(),
+            options.getContextPropagators(),
+            dataConverter);
     for (ScheduleClientInterceptor clientInterceptor : interceptors) {
       scheduleClientInvoker =
           clientInterceptor.scheduleClientCallsInterceptor(scheduleClientInvoker);

temporal-sdk/src/main/java/io/temporal/common/converter/DefaultDataConverter.java

@@ -1,15 +1,21 @@
 package io.temporal.common.converter;
 
 import com.google.common.base.Preconditions;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter;
+import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter.TransferType;
+import io.temporal.payload.context.SerializationContext;
+import java.lang.reflect.Type;
 import java.util.*;
 import javax.annotation.Nonnull;
 
 /**
  * A {@link DataConverter} that delegates payload conversion to type specific {@link
  * PayloadConverter} instances, and delegates failure conversions to a {@link FailureConverter}.
+ * Top-level values annotated with {@link TransferTypeConvertible} are converted to their transfer
+ * types before payload conversion, including when this converter is used directly.
  */
 public class DefaultDataConverter extends PayloadAndFailureDataConverter {
-
   // Order is important as the first converter that can convert the payload is used. Needs to match
   // the other SDKs. Go SDK:
   // https://github.com/temporalio/sdk-go/blob/5e5645f0c550dcf717c095ae32c76a7087d2e985/converter/default_data_converter.go#L28
@@ -62,6 +68,36 @@ public DefaultDataConverter(PayloadConverter... converters) {
     super(Arrays.asList(converters));
   }
 
+  private DefaultDataConverter(
+      List<PayloadConverter> converters,
+      Map<String, PayloadConverter> convertersMap,
+      FailureConverter failureConverter,
+      SerializationContext context) {
+    super(converters, convertersMap, failureConverter, context);
+  }
+
+  @Override
+  public <T> Optional<Payload> toPayload(T value) throws DataConverterException {
+    return super.toPayload(TemporalTransferTypeDataConverter.toTransferValue(value));
+  }
+
+  @Override
+  public <T> T fromPayload(Payload payload, Class<T> valueClass, Type valueType)
+      throws DataConverterException {
+    TransferType transfer =
+        TemporalTransferTypeDataConverter.transferTypeFor(valueClass, valueType);
+    if (transfer == null) {
+      return super.fromPayload(payload, valueClass, valueType);
+    }
+    Object decoded = super.fromPayload(payload, transfer.getRawType(), transfer.getType());
+    return transfer.fromValue(decoded, valueClass);
+  }
+
+  @Override
+  public DataConverter withContext(@Nonnull SerializationContext context) {
+    return new DefaultDataConverter(converters, convertersMap, failureConverter, context);
+  }
+
   /**
    * Modifies this {@code DefaultDataConverter} by overriding some of its {@link PayloadConverter}s.
    * Every payload converter from {@code overrideConverters} either replaces existing payload

temporal-sdk/src/main/java/io/temporal/common/converter/PayloadAndFailureDataConverter.java

@@ -8,6 +8,7 @@
 import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.failure.v1.Failure;
 import io.temporal.failure.DefaultFailureConverter;
+import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter;
 import io.temporal.payload.context.SerializationContext;
 import java.lang.reflect.Type;
 import java.util.*;
@@ -130,7 +131,7 @@ public RuntimeException failureToException(@Nonnull Failure failure) {
     return (serializationContext != null
             ? failureConverter.withContext(serializationContext)
             : failureConverter)
-        .failureToException(failure, this);
+        .failureToException(failure, TemporalTransferTypeDataConverter.wrap(this));
   }
 
   @Override
@@ -140,7 +141,7 @@ public Failure exceptionToFailure(@Nonnull Throwable throwable) {
     return (serializationContext != null
             ? failureConverter.withContext(serializationContext)
             : failureConverter)
-        .exceptionToFailure(throwable, this);
+        .exceptionToFailure(throwable, TemporalTransferTypeDataConverter.wrap(this));
   }
 
   @Override

temporal-sdk/src/main/java/io/temporal/common/converter/TransferTypeConverter.java

@@ -0,0 +1,38 @@
+package io.temporal.common.converter;
+
+import io.temporal.common.Experimental;
+import java.lang.reflect.Type;
+import javax.annotation.Nullable;
+
+/**
+ * Converts a model to and from a representation handled by the configured {@link DataConverter}.
+ *
+ * <p>Conversion is applied only to top-level values. The {@code valueType} arguments contain the
+ * complete requested model type, including generic arguments. Implementations must be stateless and
+ * thread-safe because converter instances are cached.
+ *
+ * @param <T> annotated model type converted to and from its transfer representation
+ */
+@Experimental
+public interface TransferTypeConverter<T> {
+  /**
+   * Returns the type used to serialize a model value.
+   *
+   * @param valueType complete declared model type, including generic arguments
+   * @return non-null transfer type
+   */
+  Type getTransferType(Type valueType);
+
+  /** Converts a model value to its transfer representation, which may be null. */
+  @Nullable
+  Object toTransferType(T value);
+
+  /**
+   * Reconstructs a model value from its transfer representation.
+   *
+   * @param value transfer representation decoded by the configured data converter, which may be
+   *     null
+   * @param valueType complete requested model type, including generic arguments
+   */
+  T fromTransferType(@Nullable Object value, Type valueType);
+}