diff --git a/CHANGELOG.md b/CHANGELOG.md index af4bac6c19..585e6c8122 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,11 @@ to docs, or any other relevant information. ## [Unreleased] ### 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` diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java index 77bf2bd794..ff9104cd8d 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java +++ b/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.RootActivityClientInvoker; import io.temporal.internal.client.external.GenericWorkflowClientImpl; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; @@ -37,20 +39,25 @@ 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; this.metricsScope = stubs.getOptions().getMetricsScope().tagged(MetricsTag.defaultTags(options.getNamespace())); GenericWorkflowClientImpl genericClient = new GenericWorkflowClientImpl(stubs, metricsScope); - this.invoker = initializeClientInvoker(genericClient, options); + this.invoker = initializeClientInvoker(genericClient, options, internalDataConverter); this.manualActivityCompletionClientFactory = ManualActivityCompletionClientFactory.newFactory( - stubs, options.getNamespace(), options.getIdentity(), options.getDataConverter()); + stubs, 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, dataConverter); for (ActivityClientInterceptor interceptor : options.getInterceptors()) { invoker = interceptor.activityClientCallsInterceptor(invoker); } diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java index 20aef9d887..39fbb90829 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityCompletionClientImpl.java @@ -129,7 +129,7 @@ public void reportCancellationStandalone( @Override public void heartbeat(byte[] taskToken, V details) throws ActivityCompletionException { - factory.getClient(taskToken, metricsScope).recordHeartbeat(details); + factory.getClient(taskToken, metricsScope, serializationContext).recordHeartbeat(details); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java index a9248e80b1..171b838c83 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java +++ b/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; diff --git a/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java b/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java index 414cd3c568..ba4d5d67b0 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/NexusClientOptions.java +++ b/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. */ diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java index a3b92aa219..e4d98bf5c0 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java +++ b/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,11 @@ 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 = + WorkflowClientOptions.newBuilder(options).setDataConverter(internalDataConverter).build(); workflowServiceStubs = new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace()); - this.options = options; this.workflowServiceStubs = workflowServiceStubs; this.metricsScope = workflowServiceStubs @@ -124,7 +129,7 @@ public static WorkflowClient newInstance( workflowServiceStubs, options.getNamespace(), options.getIdentity(), - options.getDataConverter(), + getInternalDataConverter(), externalStorageRunner); java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval(); @@ -160,13 +165,18 @@ public WorkflowClientOptions getOptions() { return options; } + @Override + public DataConverter getInternalDataConverter() { + return internalDataConverter; + } + @Override @SuppressWarnings("unchecked") public T newWorkflowStub(Class 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 +253,7 @@ public T newWorkflowStub( WorkflowInvocationHandler invocationHandler = new WorkflowInvocationHandler( workflowInterface, - this.getOptions(), + options, workflowClientCallsInvoker, execution.build(), legacyTargeting, diff --git a/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java b/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java index 9f96afe652..b95b66744c 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/schedules/ScheduleClientImpl.java +++ b/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,9 @@ 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, dataConverter); for (ScheduleClientInterceptor clientInterceptor : interceptors) { scheduleClientInvoker = clientInterceptor.scheduleClientCallsInterceptor(scheduleClientInvoker); diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/PayloadAndFailureDataConverter.java b/temporal-sdk/src/main/java/io/temporal/common/converter/PayloadAndFailureDataConverter.java index 935fd8462c..e29c2c35ab 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/converter/PayloadAndFailureDataConverter.java +++ b/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 diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/TransferTypeConverter.java b/temporal-sdk/src/main/java/io/temporal/common/converter/TransferTypeConverter.java new file mode 100644 index 0000000000..ae1735d562 --- /dev/null +++ b/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}. + * + *

Conversion is applied only to top-level values and performs one transfer step. 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 annotated model type converted to and from its transfer representation + */ +@Experimental +public interface TransferTypeConverter { + /** + * 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); +} diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/TransferTypeConvertible.java b/temporal-sdk/src/main/java/io/temporal/common/converter/TransferTypeConvertible.java new file mode 100644 index 0000000000..05553b72fe --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/common/converter/TransferTypeConvertible.java @@ -0,0 +1,27 @@ +package io.temporal.common.converter; + +import io.temporal.common.Experimental; +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Associates a model class with the converter used to produce its transfer representation. + * + *

The annotation is read only from the exact declared class and is not inherited. Conversion is + * applied to top-level values only and performs one transfer step. The configured {@link + * DataConverter} remains responsible for serialization, payload codecs, and wire encoding. + * + *

The converter class must be concrete and have a public no-argument constructor. Converter + * instances are cached, so implementations must be stateless and thread-safe. + */ +@Experimental +@Documented +@Retention(RetentionPolicy.RUNTIME) +@Target(ElementType.TYPE) +public @interface TransferTypeConvertible { + /** The converter associated with the annotated model class. */ + Class> value(); +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java new file mode 100644 index 0000000000..eaddfcaba6 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ClientDataConverterFactory.java @@ -0,0 +1,59 @@ +package io.temporal.internal.client; + +import com.google.common.base.Strings; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.DataConverter; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import java.util.Objects; +import javax.annotation.Nullable; + +/** Assembles client converters and scopes workflow conversion when workflow identity is known. */ +public final class ClientDataConverterFactory { + + private final String namespace; + private final ExternalStorageDataConverter workflowConverter; + + ClientDataConverterFactory( + WorkflowClientOptions clientOptions, @Nullable ExternalStorageRunner externalStorage) { + this.namespace = clientOptions.getNamespace(); + // Workflow reads must reject storage references even when no driver is configured. + this.workflowConverter = withStorage(clientOptions.getDataConverter(), externalStorage); + } + + /** Returns a transfer-aware client converter without external storage. */ + public static DataConverter forClient(DataConverter configured) { + return transferAware(configured); + } + + /** Returns a client converter with external storage outside transfer conversion. */ + public static DataConverter forClient( + DataConverter configured, ExternalStorageRunner externalStorage) { + return withStorage(configured, Objects.requireNonNull(externalStorage, "externalStorage")); + } + + private static ExternalStorageDataConverter withStorage( + DataConverter configured, @Nullable ExternalStorageRunner externalStorage) { + return new ExternalStorageDataConverter(transferAware(configured), externalStorage); + } + + private static DataConverter transferAware(DataConverter configured) { + return TemporalTransferTypeDataConverter.wrap(configured); + } + + DataConverter forWorkflow( + String workflowId, @Nullable String runId, @Nullable String workflowType) { + DataConverter converter = + workflowConverter.withContext(new WorkflowSerializationContext(namespace, workflowId)); + return ((ExternalStorageDataConverter) converter) + .withStorageTarget( + new StorageDriverWorkflowInfo( + namespace, + Strings.emptyToNull(workflowId), + Strings.emptyToNull(runId), + Strings.emptyToNull(workflowType))); + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 55297194fe..5581813159 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -49,11 +49,15 @@ public class RootActivityClientInvoker implements ActivityClientCallsInterceptor private final GenericWorkflowClient genericClient; private final ActivityClientOptions clientOptions; + private final DataConverter dataConverter; public RootActivityClientInvoker( - GenericWorkflowClient genericClient, ActivityClientOptions clientOptions) { + GenericWorkflowClient genericClient, + ActivityClientOptions clientOptions, + DataConverter dataConverter) { this.genericClient = genericClient; this.clientOptions = clientOptions; + this.dataConverter = dataConverter; } @Override @@ -63,16 +67,14 @@ public StartActivityOutput startActivity(StartActivityInput input) { throw new IllegalArgumentException("taskQueue must not be null or empty"); } DataConverter dc = - clientOptions - .getDataConverter() - .withContext( - new ActivitySerializationContext( - clientOptions.getNamespace(), - null, - null, - input.getActivityType(), - options.getTaskQueue(), - false)); + dataConverter.withContext( + new ActivitySerializationContext( + clientOptions.getNamespace(), + null, + null, + input.getActivityType(), + options.getTaskQueue(), + false)); InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null; @@ -213,8 +215,7 @@ public StartActivityOutput startActivity(StartActivityInput input) { public GetActivityResultOutput getActivityResult(GetActivityResultInput input) throws TimeoutException { String namespace = clientOptions.getNamespace(); - DataConverter dc = - clientOptions.getDataConverter().withContext(resultSerializationContext(input)); + DataConverter dc = dataConverter.withContext(resultSerializationContext(input)); Deadline deadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit()); while (true) { @@ -289,8 +290,7 @@ public GetActivityResultOutput getActivityResult(GetActivityResultInput CompletableFuture> getActivityResultAsync( GetActivityResultInput input) { - DataConverter dc = - clientOptions.getDataConverter().withContext(resultSerializationContext(input)); + DataConverter dc = dataConverter.withContext(resultSerializationContext(input)); Deadline deadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit()); return pollActivityUntilOutcome(input, deadline) .handle( @@ -394,8 +394,7 @@ public DescribeActivityOutput describeActivity(DescribeActivityInput input) { } DescribeActivityExecutionResponse response = genericClient.describeActivity(req.build()); return new DescribeActivityOutput( - new ActivityExecutionDescription( - response, clientOptions.getDataConverter(), clientOptions.getNamespace())); + new ActivityExecutionDescription(response, dataConverter, clientOptions.getNamespace())); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootScheduleClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootScheduleClientInvoker.java index ff2c5fde90..7171c52acf 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootScheduleClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootScheduleClientInvoker.java @@ -11,6 +11,7 @@ import io.temporal.api.workflowservice.v1.*; import io.temporal.client.ListScheduleListDescriptionIterator; import io.temporal.client.schedules.*; +import io.temporal.common.converter.DataConverter; import io.temporal.common.interceptors.ScheduleClientCallsInterceptor; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.common.ProtobufTimeUtils; @@ -26,14 +27,18 @@ public class RootScheduleClientInvoker implements ScheduleClientCallsInterceptor private final GenericWorkflowClient genericClient; private final ScheduleClientOptions clientOptions; + private final DataConverter dataConverter; private final ScheduleProtoUtil scheduleRequestHeader; public RootScheduleClientInvoker( - GenericWorkflowClient genericClient, ScheduleClientOptions clientOptions) { + GenericWorkflowClient genericClient, + ScheduleClientOptions clientOptions, + DataConverter dataConverter) { this.genericClient = genericClient; this.clientOptions = clientOptions; - this.scheduleRequestHeader = new ScheduleProtoUtil(genericClient, clientOptions); + this.dataConverter = dataConverter; + this.scheduleRequestHeader = new ScheduleProtoUtil(genericClient, clientOptions, dataConverter); } @Override @@ -52,8 +57,7 @@ public void createSchedule(CreateScheduleInput input) { // TODO we don't have a workflow context here, maybe we need a schedule context? request.setMemo( Memo.newBuilder() - .putAllFields( - intoPayloadMap(clientOptions.getDataConverter(), input.getOptions().getMemo()))); + .putAllFields(intoPayloadMap(dataConverter, input.getOptions().getMemo()))); } if (input.getOptions().getSearchAttributes() != null @@ -187,7 +191,7 @@ public DescribeScheduleOutput describeSchedule(DescribeScheduleInput input) { SearchAttributesUtil.decode(response.getSearchAttributes())), SearchAttributesUtil.decodeTyped(response.getSearchAttributes()), response.getMemo().getFieldsMap(), - clientOptions.getDataConverter())); + dataConverter)); } catch (Exception e) { throw new ScheduleException(e); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index d3b3321ebe..da1e920c4f 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java @@ -57,7 +57,7 @@ public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor private final QueryRejectCondition queryRejectCondition; private final EagerWorkflowTaskDispatcher eagerWorkflowTaskDispatcher; private final WorkflowClientRequestFactory requestsHelper; - private final WorkflowClientDataConverterFactory converterFactory; + private final ClientDataConverterFactory converterFactory; private final @Nullable ExternalStorageRunner externalStorage; public RootWorkflowClientInvoker( @@ -72,7 +72,7 @@ public RootWorkflowClientInvoker( WorkflowClientOptions clientOptions, WorkerFactoryRegistry workerFactoryRegistry, @Nullable ExternalStorageRunner externalStorage) { - this.converterFactory = new WorkflowClientDataConverterFactory(clientOptions, externalStorage); + this.converterFactory = new ClientDataConverterFactory(clientOptions, externalStorage); this.externalStorage = externalStorage; this.genericClient = genericClient; this.namespace = clientOptions.getNamespace(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java index 2ae576c77c..b6714cf462 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ScheduleProtoUtil.java @@ -39,11 +39,15 @@ public class ScheduleProtoUtil { private final GenericWorkflowClient genericClient; private final ScheduleClientOptions clientOptions; + private final DataConverter dataConverter; public ScheduleProtoUtil( - GenericWorkflowClient genericClient, ScheduleClientOptions clientOptions) { + GenericWorkflowClient genericClient, + ScheduleClientOptions clientOptions, + DataConverter dataConverter) { this.genericClient = genericClient; this.clientOptions = clientOptions; + this.dataConverter = dataConverter; } private io.temporal.common.interceptors.Header extractContextsAndConvertToBytes( @@ -70,12 +74,9 @@ public ScheduleAction actionToProto(io.temporal.client.schedules.ScheduleAction if (action instanceof ScheduleActionStartWorkflow) { ScheduleActionStartWorkflow startWorkflowAction = (ScheduleActionStartWorkflow) action; DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), - startWorkflowAction.getOptions().getWorkflowId())); + dataConverter.withContext( + new WorkflowSerializationContext( + clientOptions.getNamespace(), startWorkflowAction.getOptions().getWorkflowId())); WorkflowOptions wfOptions = startWorkflowAction.getOptions(); // Disallow some options @@ -401,7 +402,7 @@ public ScheduleListDescription protoToScheduleListDescription(ScheduleListEntry new ScheduleListSchedule(action, spec, state), info, entry.getMemo().getFieldsMap(), - this.clientOptions.getDataConverter(), + dataConverter, Collections.unmodifiableMap(SearchAttributesUtil.decode(entry.getSearchAttributes()))); } @@ -411,11 +412,9 @@ public io.temporal.client.schedules.ScheduleAction protoToAction(@Nonnull Schedu if (action.hasStartWorkflow()) { NewWorkflowExecutionInfo startWfAction = action.getStartWorkflow(); DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), startWfAction.getWorkflowId())); + dataConverter.withContext( + new WorkflowSerializationContext( + clientOptions.getNamespace(), startWfAction.getWorkflowId())); ScheduleActionStartWorkflow.Builder builder = ScheduleActionStartWorkflow.newBuilder(); builder.setWorkflowType(startWfAction.getWorkflowType().getName()); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java deleted file mode 100644 index 9384be6e00..0000000000 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java +++ /dev/null @@ -1,37 +0,0 @@ -package io.temporal.internal.client; - -import com.google.common.base.Strings; -import io.temporal.client.WorkflowClientOptions; -import io.temporal.common.converter.DataConverter; -import io.temporal.internal.payload.storage.ExternalStorageDataConverter; -import io.temporal.internal.payload.storage.ExternalStorageRunner; -import io.temporal.payload.context.WorkflowSerializationContext; -import io.temporal.payload.storage.StorageDriverWorkflowInfo; -import javax.annotation.Nullable; - -/** Supplies a {@link DataConverter} for clients. */ -final class WorkflowClientDataConverterFactory { - - private final String namespace; - private final DataConverter baseConverter; - - WorkflowClientDataConverterFactory( - WorkflowClientOptions clientOptions, @Nullable ExternalStorageRunner externalStorage) { - this.namespace = clientOptions.getNamespace(); - this.baseConverter = - new ExternalStorageDataConverter(clientOptions.getDataConverter(), externalStorage); - } - - DataConverter forWorkflow( - String workflowId, @Nullable String runId, @Nullable String workflowType) { - DataConverter converter = - baseConverter.withContext(new WorkflowSerializationContext(namespace, workflowId)); - return ((ExternalStorageDataConverter) converter) - .withStorageTarget( - new StorageDriverWorkflowInfo( - namespace, - Strings.emptyToNull(workflowId), - Strings.emptyToNull(runId), - Strings.emptyToNull(workflowType))); - } -} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java index 90017cd575..d26a64c253 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java @@ -2,6 +2,7 @@ import io.temporal.api.worker.v1.EnvironmentInfo; import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.DataConverter; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.worker.HeartbeatManager; import io.temporal.worker.WorkerFactory; @@ -37,4 +38,7 @@ public interface WorkflowClientInternal { @Nullable ExternalStorageRunner getExternalStorageRunner(); + + /** Returns the SDK's converter, including internal transfer-type conversion. */ + DataConverter getInternalDataConverter(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java b/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java new file mode 100644 index 0000000000..d543a0d044 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java @@ -0,0 +1,297 @@ +package io.temporal.internal.common.converter; + +import com.google.common.reflect.TypeToken; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.failure.v1.Failure; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterException; +import io.temporal.common.converter.RawValue; +import io.temporal.common.converter.TransferTypeConverter; +import io.temporal.common.converter.TransferTypeConvertible; +import io.temporal.payload.context.SerializationContext; +import java.lang.reflect.Constructor; +import java.lang.reflect.Modifier; +import java.lang.reflect.Type; +import java.util.Arrays; +import java.util.Optional; +import javax.annotation.Nonnull; + +/** Applies type-owned transfer conversion around an SDK-managed data converter. */ +public final class TemporalTransferTypeDataConverter implements DataConverter { + private static final ClassValue CONVERTER_DESCRIPTORS = + new ClassValue() { + @Override + protected ConverterDescriptor computeValue(Class type) { + TransferTypeConvertible annotation = + type.getDeclaredAnnotation(TransferTypeConvertible.class); + return annotation == null + ? ConverterDescriptor.NONE + : new ConverterDescriptor(type, annotation.value()); + } + }; + + private final DataConverter delegate; + + private TemporalTransferTypeDataConverter(DataConverter delegate) { + this.delegate = delegate; + } + + /** Wraps a converter once. */ + public static DataConverter wrap(DataConverter converter) { + if (converter instanceof TemporalTransferTypeDataConverter) { + return converter; + } + return new TemporalTransferTypeDataConverter(converter); + } + + @Override + public Optional toPayload(T value) throws DataConverterException { + return delegate.toPayload(toTransferValue(value)); + } + + @Override + public Optional toPayloads(Object... values) throws DataConverterException { + if (values == null) { + return delegate.toPayloads(values); + } + Object[] transferred = new Object[values.length]; + for (int i = 0; i < values.length; i++) { + transferred[i] = toTransferValue(values[i]); + } + return delegate.toPayloads(transferred); + } + + @Override + public T fromPayload(Payload payload, Class valueClass, Type valueType) + throws DataConverterException { + ConverterDescriptor descriptor = descriptorFor(valueClass); + if (descriptor == null) { + return delegate.fromPayload(payload, valueClass, valueType); + } + Type requestedType = requestedType(valueClass, valueType); + TransferType transfer = descriptor.transferTypeFor(requestedType); + Object value = delegate.fromPayload(payload, transfer.rawType, transfer.type); + return fromTransferValue(value, descriptor, requestedType, valueClass); + } + + @Override + public T fromPayloads( + int index, Optional content, Class valueClass, Type valueType) + throws DataConverterException { + ConverterDescriptor descriptor = descriptorFor(valueClass); + if (descriptor == null || !hasPayload(index, content)) { + return delegate.fromPayloads(index, content, valueClass, valueType); + } + Type requestedType = requestedType(valueClass, valueType); + TransferType transfer = descriptor.transferTypeFor(requestedType); + Object value = delegate.fromPayloads(index, content, transfer.rawType, transfer.type); + return fromTransferValue(value, descriptor, requestedType, valueClass); + } + + @Override + public Object[] fromPayloads( + Optional content, Class[] parameterTypes, Type[] genericParameterTypes) + throws DataConverterException { + if (parameterTypes != null + && (genericParameterTypes == null + || parameterTypes.length != genericParameterTypes.length)) { + throw new IllegalArgumentException( + "parameterTypes don't match length of valueTypes: " + + Arrays.toString(parameterTypes) + + "<>" + + Arrays.toString(genericParameterTypes)); + } + if (!content.isPresent() || content.get().getPayloadsCount() == 0) { + return delegate.fromPayloads(content, parameterTypes, genericParameterTypes); + } + + Class[] transferClasses = parameterTypes.clone(); + Type[] transferTypes = genericParameterTypes.clone(); + ConverterDescriptor[] descriptors = new ConverterDescriptor[parameterTypes.length]; + int payloadCount = content.get().getPayloadsCount(); + + for (int i = 0; i < parameterTypes.length; i++) { + ConverterDescriptor descriptor = descriptorFor(parameterTypes[i]); + descriptors[i] = descriptor; + if (descriptor != null && i < payloadCount) { + Type requestedType = requestedType(parameterTypes[i], genericParameterTypes[i]); + TransferType transfer = descriptor.transferTypeFor(requestedType); + transferClasses[i] = transfer.rawType; + transferTypes[i] = transfer.type; + } + } + + Object[] values = delegate.fromPayloads(content, transferClasses, transferTypes); + for (int i = 0; i < values.length && i < payloadCount; i++) { + if (descriptors[i] != null) { + Type requestedType = requestedType(parameterTypes[i], genericParameterTypes[i]); + values[i] = fromTransferValue(values[i], descriptors[i], requestedType, parameterTypes[i]); + } + } + return values; + } + + @Nonnull + @Override + public RuntimeException failureToException(@Nonnull Failure failure) { + return delegate.failureToException(failure); + } + + @Nonnull + @Override + public Failure exceptionToFailure(@Nonnull Throwable throwable) { + return delegate.exceptionToFailure(throwable); + } + + @Nonnull + @Override + public DataConverter withContext(@Nonnull SerializationContext context) { + return wrap(delegate.withContext(context)); + } + + private static Object toTransferValue(Object value) { + if (value == null || value instanceof RawValue) { + return value; + } + ConverterDescriptor descriptor = descriptorFor(value.getClass()); + return descriptor == null ? value : descriptor.converter().toTransferType(value); + } + + private static T fromTransferValue( + Object value, ConverterDescriptor descriptor, Type valueType, Class valueClass) { + Object result = descriptor.converter().fromTransferType(value, valueType); + if (result != null && !valueClass.isInstance(result)) { + throw new DataConverterException( + "Transfer type converter " + + descriptor.converterClass.getName() + + " declared by " + + descriptor.modelClass.getName() + + " returned " + + result.getClass().getName() + + " instead of " + + valueClass.getName()); + } + return valueClass.cast(result); + } + + private static Type requestedType(Class valueClass, Type valueType) { + return valueType == null ? valueClass : valueType; + } + + private static ConverterDescriptor descriptorFor(Class valueClass) { + if (valueClass == RawValue.class) { + return null; + } + ConverterDescriptor descriptor = CONVERTER_DESCRIPTORS.get(valueClass); + return descriptor == ConverterDescriptor.NONE ? null : descriptor; + } + + private static boolean hasPayload(int index, Optional content) { + return content.isPresent() && index >= 0 && index < content.get().getPayloadsCount(); + } + + private static final class TransferType { + private final Type type; + private final Class rawType; + + @SuppressWarnings("unchecked") + private TransferType(Type type) { + this.type = type; + this.rawType = (Class) TypeToken.of(type).getRawType(); + } + } + + /** + * Owns the transfer converter declaration and its lazily initialized converter instance for a + * model class. + * + *

The {@link ClassValue} cache publishes one descriptor to callers before converter + * construction. Synchronizing initialization on that shared descriptor prevents concurrent + * callers from constructing duplicate converter instances. Once construction succeeds, every + * caller reuses the same instance. + */ + private static final class ConverterDescriptor { + /** Represents cached absence while keeping descriptor lookup results non-null. */ + private static final ConverterDescriptor NONE = new ConverterDescriptor(); + + private final Class modelClass; + private final Class> converterClass; + private volatile TransferTypeConverter converter; + + private ConverterDescriptor() { + this.modelClass = null; + this.converterClass = null; + } + + private ConverterDescriptor( + Class modelClass, Class> converterClass) { + this.modelClass = modelClass; + this.converterClass = converterClass; + } + + private TransferType transferTypeFor(Type valueType) { + Type type = converter().getTransferType(valueType); + // The delegate and TypeToken require a concrete type to decode the transfer representation. + if (type == null) { + throw new DataConverterException( + "Transfer type converter " + + converterClass.getName() + + " returned a null transfer type for " + + modelClass.getName()); + } + return new TransferType(type); + } + + /** + * Returns the shared converter, constructing it on first use. The volatile fast path avoids + * synchronization after the instance has been safely published. + */ + @SuppressWarnings("unchecked") + private TransferTypeConverter converter() { + TransferTypeConverter result = converter; + if (result != null) { + return result; + } + synchronized (this) { + if (converter == null) { + converter = (TransferTypeConverter) instantiateConverter(); + } + return converter; + } + } + + private TransferTypeConverter instantiateConverter() { + // Interfaces and abstract classes cannot provide the converter instance used at runtime. + if (converterClass.isInterface() || Modifier.isAbstract(converterClass.getModifiers())) { + throw declarationFailure("must be a concrete class", null); + } + // A non-static member class requires an enclosing instance that the SDK does not own. + if (converterClass.isMemberClass() && !Modifier.isStatic(converterClass.getModifiers())) { + throw declarationFailure("must be static when declared as an inner class", null); + } + try { + // A public no-argument constructor keeps converter creation independent of application + // state. + Constructor> constructor = + converterClass.getConstructor(); + return constructor.newInstance(); + } catch (ReflectiveOperationException | SecurityException e) { + throw declarationFailure("must have an accessible public no-argument constructor", e); + } + } + + private DataConverterException declarationFailure(String reason, Throwable cause) { + String message = + "Invalid transfer type converter " + + converterClass.getName() + + " declared by " + + modelClass.getName() + + ": " + + reason; + return cause == null + ? new DataConverterException(message) + : new DataConverterException(message, cause); + } + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 91f5515ea8..b09827f838 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -130,10 +130,13 @@ private static final class TaskSnapshot { this.options = WorkerOptions.newBuilder(options).validateAndBuildWithDefaults(); this.clientOptions = client.getOptions(); this.cache = cache; - ExternalStorageRunner externalStorageRunner = - ((WorkflowClientInternal) client.getInternal()).getExternalStorageRunner(); + WorkflowClientInternal clientInternal = (WorkflowClientInternal) client.getInternal(); + ExternalStorageRunner externalStorageRunner = clientInternal.getExternalStorageRunner(); factoryOptions = WorkerFactoryOptions.newBuilder(factoryOptions).validateAndBuildWithDefaults(); - WorkflowClientOptions clientOptions = client.getOptions(); + WorkflowClientOptions clientOptions = + WorkflowClientOptions.newBuilder(client.getOptions()) + .setDataConverter(clientInternal.getInternalDataConverter()) + .build(); String namespace = clientOptions.getNamespace(); this.namespace = namespace; String workerControlTaskQueue = diff --git a/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java index 162ea7f8ca..12f45a7745 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/ActivityCompletionClientImplTest.java @@ -9,6 +9,7 @@ import io.temporal.activity.ManualActivityCompletionClient; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.workflow.Functions; import java.util.Optional; import org.junit.Before; @@ -69,6 +70,20 @@ public void testCompleteStandaloneCallsCompletionHandleOnException() { verify(completionHandle).apply(); } + @Test + public void taskTokenHeartbeatUsesSerializationContext() { + byte[] taskToken = new byte[] {1, 2, 3}; + ActivitySerializationContext context = + new ActivitySerializationContext( + "test-namespace", "workflow-id", "workflow-type", "activity-type", "queue", false); + when(factory.getClient(taskToken, scope, context)).thenReturn(manualClient); + + client.withContext(context).heartbeat(taskToken, "details"); + + verify(factory).getClient(taskToken, scope, context); + verify(manualClient).recordHeartbeat("details"); + } + // completeExceptionally(String activityId, Optional activityRunId, Exception result) @Test diff --git a/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java index 27f1ee5be8..ea7ef92986 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/NexusClientOptionsTest.java @@ -6,7 +6,6 @@ import io.temporal.api.common.v1.Payload; import io.temporal.common.converter.DataConverter; import io.temporal.common.interceptors.NexusClientInterceptor; -import io.temporal.internal.payload.storage.ExternalStorageDataConverter; import io.temporal.payload.storage.ExternalStorage; import io.temporal.payload.storage.StorageDriver; import io.temporal.payload.storage.StorageDriverClaim; @@ -71,8 +70,9 @@ public void externalStorageSurvivesBuild() { } @Test - public void resolveOptionsWrapsDataConverterWithoutChangingConfiguredOptions() { + public void resolveOptionsUsesInternalConverterWithoutChangingConfiguredOptions() { DataConverter dataConverter = mock(DataConverter.class); + DataConverter internalDataConverter = mock(DataConverter.class); NexusClientOptions options = NexusClientOptions.newBuilder() .setDataConverter(dataConverter) @@ -80,8 +80,8 @@ public void resolveOptionsWrapsDataConverterWithoutChangingConfiguredOptions() { .build(); assertSame(dataConverter, options.getDataConverter()); - assertTrue( - options.toResolvedOptions().getDataConverter() instanceof ExternalStorageDataConverter); + assertSame( + internalDataConverter, options.toResolvedOptions(internalDataConverter).getDataConverter()); } private static ExternalStorage storage() { diff --git a/temporal-sdk/src/test/java/io/temporal/client/functional/ActivityClientTransferTypeTest.java b/temporal-sdk/src/test/java/io/temporal/client/functional/ActivityClientTransferTypeTest.java new file mode 100644 index 0000000000..62ef54f506 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/functional/ActivityClientTransferTypeTest.java @@ -0,0 +1,69 @@ +package io.temporal.client.functional; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.StartActivityOptions; +import io.temporal.common.converter.TransferTypeTestModel; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.time.Duration; +import java.util.UUID; +import org.junit.Rule; +import org.junit.Test; + +public class ActivityClientTransferTypeTest { + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setActivityImplementations(new TransferActivityImpl()) + .build(); + + @Test + public void activityClientAndWorkerRoundTripTransferTypes() { + assumeTrue( + "server does not support standalone activities", SDKTestWorkflowRule.useExternalService); + ActivityClient client = + ActivityClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .build()); + StartActivityOptions options = + StartActivityOptions.newBuilder() + .setId(UUID.randomUUID().toString()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setScheduleToCloseTimeout(Duration.ofMinutes(1)) + .build(); + + TransferTypeTestModel result = + client.execute( + TransferActivity.class, + TransferActivity::execute, + options, + new TransferTypeTestModel("input")); + + assertEquals(new TransferTypeTestModel("input-activity"), result); + assertTrue(result.wasTransferred()); + } + + @ActivityInterface + public interface TransferActivity { + @ActivityMethod + TransferTypeTestModel execute(TransferTypeTestModel input); + } + + public static final class TransferActivityImpl implements TransferActivity { + @Override + public TransferTypeTestModel execute(TransferTypeTestModel input) { + if (!input.wasTransferred()) { + throw new IllegalStateException("Activity input did not use its transfer type converter"); + } + return new TransferTypeTestModel(input.value() + "-activity"); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/client/nexus/NexusServiceClientTest.java b/temporal-sdk/src/test/java/io/temporal/client/nexus/NexusServiceClientTest.java index c880a9bd01..9b8476e8f1 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/nexus/NexusServiceClientTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/nexus/NexusServiceClientTest.java @@ -2,6 +2,8 @@ import static org.junit.Assume.assumeTrue; +import io.nexusrpc.Operation; +import io.nexusrpc.Service; import io.nexusrpc.handler.OperationHandler; import io.nexusrpc.handler.OperationImpl; import io.nexusrpc.handler.ServiceImpl; @@ -13,6 +15,7 @@ import io.temporal.client.NexusServiceClient; import io.temporal.client.StartNexusOperationOptions; import io.temporal.client.UntypedNexusOperationHandle; +import io.temporal.common.converter.TransferTypeTestModel; import io.temporal.testing.CloudTestExclusion.RequiresCloudProvisioning; import io.temporal.testing.CloudTestExclusionNote; import io.temporal.testing.internal.SDKTestWorkflowRule; @@ -42,6 +45,7 @@ public class NexusServiceClientTest { .setWorkflowTypes(PlaceholderWorkflowImpl.class) .setNexusServiceImplementation( new EchoNexusServiceImpl(), + new TransferServiceImpl(), new VoidInputServiceImpl(), new VoidReturnServiceImpl(), new VoidServiceImpl()) @@ -65,6 +69,18 @@ public void executeReturnsTypedResult() { Assert.assertEquals("echo:hello", result); } + @Test + public void executeRoundTripsTransferTypes() { + NexusServiceClient client = buildServiceClientFor(TransferService.class); + + TransferTypeTestModel result = + client.execute( + TransferService::operation, newOptionsWithId(), new TransferTypeTestModel("input")); + + Assert.assertEquals(new TransferTypeTestModel("input-nexus"), result); + Assert.assertTrue(result.wasTransferred()); + } + @Test public void startReturnsTypedHandleAndPollsResult() { NexusServiceClient client = @@ -224,6 +240,27 @@ public String execute(String input) { } } + @Service + public interface TransferService { + @Operation + TransferTypeTestModel operation(TransferTypeTestModel input); + } + + @ServiceImpl(service = TransferService.class) + public static class TransferServiceImpl { + @OperationImpl + public OperationHandler operation() { + return OperationHandler.sync( + (ctx, details, input) -> { + if (!input.wasTransferred()) { + throw new IllegalStateException( + "Nexus input did not use its transfer type converter"); + } + return new TransferTypeTestModel(input.value() + "-nexus"); + }); + } + } + /** Handler for the no-input, has-output service {@code TestNexusServiceVoidInput}. */ @ServiceImpl(service = TestNexusServices.TestNexusServiceVoidInput.class) public static class VoidInputServiceImpl { diff --git a/temporal-sdk/src/test/java/io/temporal/client/schedules/ScheduleTest.java b/temporal-sdk/src/test/java/io/temporal/client/schedules/ScheduleTest.java index c89953b1a9..b5eabc4921 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/schedules/ScheduleTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/schedules/ScheduleTest.java @@ -7,7 +7,9 @@ import io.temporal.common.RetryOptions; import io.temporal.common.SearchAttributeKey; import io.temporal.common.SearchAttributes; +import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.converter.EncodedValues; +import io.temporal.common.converter.TransferTypeTestModel; import io.temporal.common.interceptors.ScheduleClientInterceptor; import io.temporal.testUtils.Eventually; import io.temporal.testing.CloudTestExclusion.RequiresCloudProvisioning; @@ -119,6 +121,49 @@ public void createSchedule() { } } + @Test + public void scheduleArgumentsRoundTripTransferTypes() { + ScheduleClient client = + ScheduleClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ScheduleClientOptions.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .setIdentity(testWorkflowRule.getWorkflowClient().getOptions().getIdentity()) + .setDataConverter(DefaultDataConverter.newDefaultInstance()) + .build()); + String scheduleId = UUID.randomUUID().toString(); + Schedule schedule = + Schedule.newBuilder() + .setAction( + ScheduleActionStartWorkflow.newBuilder() + .setWorkflowType("TestWorkflow1") + .setArguments(new TransferTypeTestModel("scheduled")) + .setOptions( + WorkflowOptions.newBuilder() + .setWorkflowId("workflow-" + scheduleId) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()) + .build()) + .setSpec( + ScheduleSpec.newBuilder() + .setIntervals(Arrays.asList(new ScheduleIntervalSpec(Duration.ofMinutes(1)))) + .build()) + .setState(ScheduleState.newBuilder().setPaused(true).build()) + .build(); + ScheduleHandle handle = + client.createSchedule(scheduleId, schedule, ScheduleOptions.newBuilder().build()); + try { + ScheduleActionStartWorkflow action = + (ScheduleActionStartWorkflow) handle.describe().getSchedule().getAction(); + + TransferTypeTestModel argument = action.getArguments().get(0, TransferTypeTestModel.class); + Assert.assertEquals(new TransferTypeTestModel("scheduled"), argument); + Assert.assertTrue(argument.wasTransferred()); + } finally { + handle.delete(); + } + } + @Test public void pauseUnpauseSchedule() { ScheduleClient client = createScheduleClient(); diff --git a/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java index 5b3c2a7118..667b0ee354 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java @@ -22,7 +22,15 @@ import static org.junit.Assert.*; +import com.google.protobuf.StringValue; +import io.temporal.api.common.v1.Payload; +import io.temporal.client.WorkflowClient; import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.TransferTypeConverter; +import io.temporal.common.converter.TransferTypeConvertible; +import io.temporal.internal.client.WorkflowClientInternal; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.testing.CloudTestExclusion.RequiresLocalServer; import io.temporal.testing.CloudTestExclusionNote; @@ -31,6 +39,7 @@ import io.temporal.worker.WorkerFactoryOptions; import io.temporal.worker.WorkerOptions; import io.temporal.worker.WorkerPlugin; +import java.lang.reflect.Type; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -46,6 +55,68 @@ @Category(RequiresLocalServer.class) public class PluginPropagationTest { + @Test + public void workflowPluginReplacesConverterBeforeTransferWrapping() { + DataConverter originalConverter = DefaultDataConverter.newDefaultInstance(); + TransferConverterPlugin plugin = new TransferConverterPlugin(); + WorkflowClientOptions originalOptions = + WorkflowClientOptions.newBuilder() + .setDataConverter(originalConverter) + .setPlugins(plugin) + .build(); + TestEnvironmentOptions testOptions = + TestEnvironmentOptions.newBuilder().setWorkflowClientOptions(originalOptions).build(); + + TestWorkflowEnvironment env = TestWorkflowEnvironment.newInstance(testOptions); + try { + assertSame(originalConverter, originalOptions.getDataConverter()); + WorkflowClient client = env.getWorkflowClient(); + DataConverter exposed = client.getOptions().getDataConverter(); + assertNotSame(plugin.dataConverter, exposed); + assertSame( + exposed, ((WorkflowClientInternal) client.getInternal()).getInternalDataConverter()); + Payload payload = exposed.toPayload(new TransferModel()).get(); + assertEquals("json/protobuf", payload.getMetadataOrThrow("encoding").toStringUtf8()); + } finally { + env.close(); + } + } + + private static final class TransferConverterPlugin extends SimplePlugin { + private final DataConverter dataConverter = DefaultDataConverter.newDefaultInstance(); + + private TransferConverterPlugin() { + super("transfer-converter"); + } + + @Override + public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder builder) { + builder.setDataConverter(dataConverter); + } + } + + @TransferTypeConvertible(TransferModelConverter.class) + private static final class TransferModel {} + + public static final class TransferModelConverter implements TransferTypeConverter { + public TransferModelConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return StringValue.class; + } + + @Override + public Object toTransferType(TransferModel value) { + return StringValue.of("transferred"); + } + + @Override + public TransferModel fromTransferType(Object value, Type valueType) { + return new TransferModel(); + } + } + /** A plugin that tracks all configuration calls via subclassing. */ private static class TrackingPlugin extends SimplePlugin { private final List callLog; diff --git a/temporal-sdk/src/test/java/io/temporal/common/converter/CodecDataConverterTest.java b/temporal-sdk/src/test/java/io/temporal/common/converter/CodecDataConverterTest.java index f3957ada40..6b512f07c0 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/converter/CodecDataConverterTest.java +++ b/temporal-sdk/src/test/java/io/temporal/common/converter/CodecDataConverterTest.java @@ -10,6 +10,7 @@ import io.temporal.api.failure.v1.Failure; import io.temporal.failure.ApplicationFailure; import io.temporal.failure.TemporalFailure; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; import io.temporal.payload.codec.PayloadCodec; import io.temporal.payload.codec.PayloadCodecException; import java.util.Collections; @@ -109,6 +110,22 @@ public void testDetailsAreEncoded() { assertArrayEquals(new int[] {1, 2, 3}, decodedDetailsPayloads.get(2, int[].class, int[].class)); } + @Test + public void transferTypeFailureDetailsAreEncodedOnce() { + DataConverter transferAware = TemporalTransferTypeDataConverter.wrap(dataConverter); + Failure failure = + transferAware.exceptionToFailure( + ApplicationFailure.newFailure("Message", "Type", new TransferTypeTestModel("detail"))); + + Payload detail = failure.getApplicationFailureInfo().getDetails().getPayloads(0); + assertTrue(isEncoded(detail)); + ApplicationFailure restored = (ApplicationFailure) transferAware.failureToException(failure); + TransferTypeTestModel model = restored.getDetails().get(0, TransferTypeTestModel.class); + assertEquals("detail", model.value()); + assertTrue(model.wasTransferred()); + assertEquals("Message", restored.getOriginalMessage()); + } + @Test public void testRawValuePassThrough() { Payload p = Payload.newBuilder().setData(ByteString.copyFromUtf8("test")).build(); diff --git a/temporal-sdk/src/test/java/io/temporal/common/converter/TransferTypeTestModel.java b/temporal-sdk/src/test/java/io/temporal/common/converter/TransferTypeTestModel.java new file mode 100644 index 0000000000..e0de7009a2 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/common/converter/TransferTypeTestModel.java @@ -0,0 +1,58 @@ +package io.temporal.common.converter; + +import com.google.protobuf.StringValue; +import java.lang.reflect.Type; +import java.util.Objects; + +@TransferTypeConvertible(TransferTypeTestModel.Converter.class) +public final class TransferTypeTestModel { + private final String value; + private final boolean transferred; + + public TransferTypeTestModel(String value) { + this(value, false); + } + + private TransferTypeTestModel(String value, boolean transferred) { + this.value = value; + this.transferred = transferred; + } + + public String value() { + return value; + } + + public boolean wasTransferred() { + return transferred; + } + + @Override + public boolean equals(Object other) { + return other instanceof TransferTypeTestModel + && Objects.equals(value, ((TransferTypeTestModel) other).value); + } + + @Override + public int hashCode() { + return Objects.hashCode(value); + } + + public static final class Converter implements TransferTypeConverter { + public Converter() {} + + @Override + public Type getTransferType(Type valueType) { + return StringValue.class; + } + + @Override + public Object toTransferType(TransferTypeTestModel value) { + return StringValue.of(value.value); + } + + @Override + public TransferTypeTestModel fromTransferType(Object value, Type valueType) { + return new TransferTypeTestModel(((StringValue) value).getValue(), true); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/functional/serialization/TransferTypeIntegrationTest.java b/temporal-sdk/src/test/java/io/temporal/functional/serialization/TransferTypeIntegrationTest.java new file mode 100644 index 0000000000..1f7fa2e038 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/functional/serialization/TransferTypeIntegrationTest.java @@ -0,0 +1,140 @@ +package io.temporal.functional.serialization; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.ActivityOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.common.RetryOptions; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.common.converter.TransferTypeTestModel; +import io.temporal.testing.WorkflowReplayer; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import org.junit.Rule; +import org.junit.Test; + +public class TransferTypeIntegrationTest { + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TransferWorkflowImpl.class, ActivityTransferWorkflowImpl.class) + .setActivityImplementations(new TransferActivityImpl()) + .build(); + + @Test + public void workflowClientAndWorkerRoundTripTransferTypes() { + TransferWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(TransferWorkflow.class); + + TransferTypeTestModel result = workflow.execute(new TransferTypeTestModel("input")); + + assertEquals(new TransferTypeTestModel("input-workflow"), result); + assertTrue(result.wasTransferred()); + } + + @Test + public void workflowClientOptionsExposeTransferAwareConverter() { + TransferTypeTestModel model = new TransferTypeTestModel("client"); + io.temporal.common.converter.DataConverter exposed = + testWorkflowRule.getWorkflowClient().getOptions().getDataConverter(); + + io.temporal.api.common.v1.Payload payload = exposed.toPayload(model).get(); + assertEquals("json/protobuf", payload.getMetadataOrThrow("encoding").toStringUtf8()); + assertTrue( + exposed + .fromPayload(payload, TransferTypeTestModel.class, TransferTypeTestModel.class) + .wasTransferred()); + } + + @Test + public void workflowHistoryWithTransferTypesReplays() throws Exception { + TransferWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(TransferWorkflow.class); + workflow.execute(new TransferTypeTestModel("replay")); + WorkflowStub untyped = WorkflowStub.fromTyped(workflow); + WorkflowExecutionHistory history = + testWorkflowRule.getExecutionHistory( + untyped.getExecution().getWorkflowId(), untyped.getExecution().getRunId()); + + WorkflowReplayer.replayWorkflowExecution(history, testWorkflowRule.getWorker()); + } + + @Test + public void activityStubRoundTripsTransferTypes() { + ActivityTransferWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(ActivityTransferWorkflow.class); + + TransferTypeTestModel result = workflow.execute(new TransferTypeTestModel("input")); + + assertEquals(new TransferTypeTestModel("input-activity-workflow"), result); + assertTrue(result.wasTransferred()); + } + + @WorkflowInterface + public interface TransferWorkflow { + @WorkflowMethod + TransferTypeTestModel execute(TransferTypeTestModel input); + } + + public static final class TransferWorkflowImpl implements TransferWorkflow { + @Override + public TransferTypeTestModel execute(TransferTypeTestModel input) { + if (!input.wasTransferred()) { + throw new IllegalStateException("Workflow input did not use its transfer type converter"); + } + return new TransferTypeTestModel(input.value() + "-workflow"); + } + } + + @WorkflowInterface + public interface ActivityTransferWorkflow { + @WorkflowMethod + TransferTypeTestModel execute(TransferTypeTestModel input); + } + + public static final class ActivityTransferWorkflowImpl implements ActivityTransferWorkflow { + private final TransferActivity activity = + Workflow.newActivityStub(TransferActivity.class, options()); + + @Override + public TransferTypeTestModel execute(TransferTypeTestModel input) { + if (!input.wasTransferred()) { + throw new IllegalStateException("Workflow input did not use its transfer type converter"); + } + TransferTypeTestModel result = activity.execute(input); + if (!result.wasTransferred()) { + throw new IllegalStateException("Activity result did not use its transfer type converter"); + } + return new TransferTypeTestModel(result.value() + "-workflow"); + } + } + + @ActivityInterface + public interface TransferActivity { + @ActivityMethod + TransferTypeTestModel execute(TransferTypeTestModel input); + } + + public static final class TransferActivityImpl implements TransferActivity { + @Override + public TransferTypeTestModel execute(TransferTypeTestModel input) { + if (!input.wasTransferred()) { + throw new IllegalStateException("Activity input did not use its transfer type converter"); + } + return new TransferTypeTestModel(input.value() + "-activity"); + } + } + + private static ActivityOptions options() { + return ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build(); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java index 594a8472d8..1b879df824 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java @@ -40,7 +40,10 @@ public class ActivityHandleOperatorCommandsTest { private UntypedActivityHandle newHandle() { return new ActivityHandleImpl( - "act-1", "run-1", new RootActivityClientInvoker(genericClient, clientOptions)); + "act-1", + "run-1", + new RootActivityClientInvoker( + genericClient, clientOptions, clientOptions.getDataConverter())); } @Test @@ -165,7 +168,9 @@ public void omittedJitterIsLeftOffTheWire() { */ @Test public void aRepeatedKeyFromAnInterceptorIsRejected() { - RootActivityClientInvoker invoker = new RootActivityClientInvoker(genericClient, clientOptions); + RootActivityClientInvoker invoker = + new RootActivityClientInvoker( + genericClient, clientOptions, clientOptions.getDataConverter()); IllegalArgumentException e = assertThrows( diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/ClientDataConverterFactoryTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/ClientDataConverterFactoryTest.java new file mode 100644 index 0000000000..7b324e2bed --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/ClientDataConverterFactoryTest.java @@ -0,0 +1,102 @@ +package io.temporal.internal.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import io.temporal.api.common.v1.Payload; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.TransferTypeTestModel; +import io.temporal.internal.payload.storage.ExternalStorageNotConfiguredException; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.internal.payload.storage.TestStorageDriver; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import org.junit.Test; + +public class ClientDataConverterFactoryTest { + @Test + public void offloadsSerializedTransferTypeAndReconstructsModel() { + TestStorageDriver driver = TestStorageDriver.create(); + ExternalStorageRunner runner = + ExternalStorageRunner.create( + ExternalStorage.newBuilder().setDriver(driver).setPayloadSizeThreshold(0).build()); + DataConverter converter = + ClientDataConverterFactory.forClient(DefaultDataConverter.newDefaultInstance(), runner); + + Payload reference = converter.toPayload(new TransferTypeTestModel("value")).get(); + + assertEquals( + "temporal.api.sdk.v1.ExternalStorageReference", + reference.getMetadataOrThrow("messageType").toStringUtf8()); + assertEquals(1, driver.storedPayloads().size()); + Payload stored = driver.storedPayloads().iterator().next(); + assertEquals("json/protobuf", stored.getMetadataOrThrow("encoding").toStringUtf8()); + assertEquals( + "google.protobuf.StringValue", stored.getMetadataOrThrow("messageType").toStringUtf8()); + TransferTypeTestModel result = + converter.fromPayload(reference, TransferTypeTestModel.class, TransferTypeTestModel.class); + assertEquals(new TransferTypeTestModel("value"), result); + assertTrue(result.wasTransferred()); + } + + @Test + public void scopesWorkflowConverterWithoutChangingConversionOrder() { + TestStorageDriver driver = TestStorageDriver.create(); + ExternalStorageRunner runner = + ExternalStorageRunner.create( + ExternalStorage.newBuilder().setDriver(driver).setPayloadSizeThreshold(0).build()); + WorkflowClientOptions options = + WorkflowClientOptions.newBuilder() + .setNamespace("test-namespace") + .setDataConverter(DefaultDataConverter.newDefaultInstance()) + .build(); + DataConverter scoped = + new ClientDataConverterFactory(options, runner) + .forWorkflow("workflow-id", "run-id", "workflow-type"); + + Payload reference = scoped.toPayload(new TransferTypeTestModel("value")).get(); + + assertEquals( + "google.protobuf.StringValue", + driver.storedPayloads().iterator().next().getMetadataOrThrow("messageType").toStringUtf8()); + assertEquals( + new StorageDriverWorkflowInfo("test-namespace", "workflow-id", "run-id", "workflow-type"), + driver.targets.get(0)); + assertTrue( + scoped + .fromPayload(reference, TransferTypeTestModel.class, TransferTypeTestModel.class) + .wasTransferred()); + } + + @Test + public void unconfiguredWorkflowRejectsReferencesButOtherClientsRemainTransferOnly() { + TestStorageDriver driver = TestStorageDriver.create(); + ExternalStorageRunner runner = + ExternalStorageRunner.create( + ExternalStorage.newBuilder().setDriver(driver).setPayloadSizeThreshold(0).build()); + DataConverter configured = DefaultDataConverter.newDefaultInstance(); + Payload reference = + ClientDataConverterFactory.forClient(configured, runner) + .toPayload(new TransferTypeTestModel("value")) + .get(); + DataConverter unconfiguredClient = ClientDataConverterFactory.forClient(configured); + DataConverter unconfiguredWorkflow = + new ClientDataConverterFactory( + WorkflowClientOptions.newBuilder() + .setNamespace("test-namespace") + .setDataConverter(configured) + .build(), + null) + .forWorkflow("workflow-id", null, null); + + assertTrue(unconfiguredClient.toPayload(new TransferTypeTestModel("inline")).isPresent()); + assertThrows( + ExternalStorageNotConfiguredException.class, + () -> + unconfiguredWorkflow.fromPayload( + reference, TransferTypeTestModel.class, TransferTypeTestModel.class)); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java index 577286df4f..39a594dc03 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java @@ -14,6 +14,8 @@ import io.temporal.client.ActivityClientOptions; import io.temporal.client.StartActivityOptions; import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.TransferTypeTestModel; import io.temporal.common.interceptors.ActivityClientCallsInterceptor.StartActivityInput; import io.temporal.common.interceptors.Header; import io.temporal.internal.client.external.GenericWorkflowClient; @@ -55,7 +57,8 @@ public void setUp() { ActivityClientOptions.newBuilder() .setNamespace(NAMESPACE) .setIdentity("test-identity") - .build()); + .build(), + DefaultDataConverter.newDefaultInstance()); nexusContext = new InternalNexusOperationContext( NAMESPACE, @@ -242,6 +245,33 @@ public void missingTaskQueueFailsAtInvocation() { Assert.assertEquals("taskQueue must not be null or empty", exception.getMessage()); } + @Test + public void startUsesTheInternalTransferAwareConverter() { + invoker = + new RootActivityClientInvoker( + genericClient, + ActivityClientOptions.newBuilder().setNamespace(NAMESPACE).build(), + ClientDataConverterFactory.forClient(DefaultDataConverter.newDefaultInstance())); + invoker.startActivity( + new StartActivityInput( + "TestActivity", + Collections.singletonList(new TransferTypeTestModel("value")), + newStartActivityInput().getOptions(), + Header.empty())); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + Assert.assertEquals( + "google.protobuf.StringValue", + captor + .getValue() + .getInput() + .getPayloads(0) + .getMetadataOrThrow("messageType") + .toStringUtf8()); + } + private static StartActivityInput newStartActivityInput() { StartActivityOptions options = StartActivityOptions.newBuilder() diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java index d0304d5425..ee5fc1e016 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/ScheduleProtoUtilTest.java @@ -1,12 +1,19 @@ package io.temporal.internal.client; +import io.temporal.api.schedule.v1.ScheduleAction; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.schedules.ScheduleActionStartWorkflow; +import io.temporal.client.schedules.ScheduleClientOptions; import io.temporal.client.schedules.SchedulePolicy; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.TransferTypeTestModel; import java.time.Duration; import org.junit.Assert; import org.junit.Test; public class ScheduleProtoUtilTest { - private final ScheduleProtoUtil util = new ScheduleProtoUtil(null, null); + private final ScheduleProtoUtil util = new ScheduleProtoUtil(null, null, null); @Test public void policyToProtoOmitsDefaultCatchupWindow() { @@ -22,4 +29,32 @@ public void policyToProtoIncludesExplicitCatchupWindow() { .getCatchupWindow() .getSeconds()); } + + @Test + public void actionUsesInternalConverterWhileOptionsKeepConfiguredConverter() { + DataConverter configured = DefaultDataConverter.newDefaultInstance(); + ScheduleClientOptions options = + ScheduleClientOptions.newBuilder().setDataConverter(configured).build(); + ScheduleProtoUtil util = + new ScheduleProtoUtil(null, options, ClientDataConverterFactory.forClient(configured)); + ScheduleActionStartWorkflow action = + ScheduleActionStartWorkflow.newBuilder() + .setWorkflowType("workflow-type") + .setArguments(new TransferTypeTestModel("value")) + .setOptions( + WorkflowOptions.newBuilder().setWorkflowId("id").setTaskQueue("queue").build()) + .build(); + + ScheduleAction proto = util.actionToProto(action); + + Assert.assertSame(configured, options.getDataConverter()); + Assert.assertEquals( + "google.protobuf.StringValue", + proto + .getStartWorkflow() + .getInput() + .getPayloads(0) + .getMetadataOrThrow("messageType") + .toStringUtf8()); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverterTest.java new file mode 100644 index 0000000000..25ce98805a --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverterTest.java @@ -0,0 +1,813 @@ +package io.temporal.internal.common.converter; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.reflect.TypeToken; +import com.google.protobuf.StringValue; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.failure.v1.Failure; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterException; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.FailureConverter; +import io.temporal.common.converter.RawValue; +import io.temporal.common.converter.TransferTypeConverter; +import io.temporal.common.converter.TransferTypeConvertible; +import io.temporal.failure.ApplicationFailure; +import io.temporal.failure.DefaultFailureConverter; +import io.temporal.payload.context.SerializationContext; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +public class TemporalTransferTypeDataConverterTest { + private DataConverter converter; + + @Before + public void setUp() { + converter = TemporalTransferTypeDataConverter.wrap(DefaultDataConverter.newDefaultInstance()); + } + + @Test + public void protobufTransferTypeRoundTripsWithoutTransferMetadata() { + Payload payload = converter.toPayload(new Model("value")).get(); + + assertEquals("json/protobuf", payload.getMetadataOrThrow("encoding").toStringUtf8()); + assertFalse(payload.getMetadataMap().containsKey("temporal-transfer-type")); + assertEquals(new Model("value"), converter.fromPayload(payload, Model.class, Model.class)); + } + + @Test + public void failureDetailsUseTransferTypesOnBothSides() { + Failure failure = + converter.exceptionToFailure( + ApplicationFailure.newFailure("message", "type", new Model("detail"))); + + assertEquals( + "json/protobuf", + failure + .getApplicationFailureInfo() + .getDetails() + .getPayloads(0) + .getMetadataOrThrow("encoding") + .toStringUtf8()); + ApplicationFailure restored = (ApplicationFailure) converter.failureToException(failure); + assertEquals(new Model("detail"), restored.getDetails().get(0, Model.class, Model.class)); + } + + @Test + public void defaultConverterUsesTransferTypesForFailureDetails() { + DataConverter unwrapped = DefaultDataConverter.newDefaultInstance(); + Failure failure = + unwrapped.exceptionToFailure( + ApplicationFailure.newFailure("message", "type", new Model("detail"))); + + assertEquals( + "json/protobuf", + failure + .getApplicationFailureInfo() + .getDetails() + .getPayloads(0) + .getMetadataOrThrow("encoding") + .toStringUtf8()); + ApplicationFailure restored = (ApplicationFailure) unwrapped.failureToException(failure); + assertEquals(new Model("detail"), restored.getDetails().get(0, Model.class, Model.class)); + } + + @Test + public void configuredFailureConverterReceivesTransferAwareDetailsConverter() { + FailureConverter failureConverter = + new FailureConverter() { + private final DefaultFailureConverter standard = new DefaultFailureConverter(); + + @Override + public RuntimeException failureToException( + Failure failure, DataConverter detailsConverter) { + assertEquals( + new Model("detail"), + detailsConverter.fromPayload( + failure.getApplicationFailureInfo().getDetails().getPayloads(0), + Model.class, + Model.class)); + return standard.failureToException(failure, detailsConverter); + } + + @Override + public Failure exceptionToFailure(Throwable throwable, DataConverter detailsConverter) { + assertEquals( + "json/protobuf", + detailsConverter + .toPayload(new Model("detail")) + .get() + .getMetadataOrThrow("encoding") + .toStringUtf8()); + return standard.exceptionToFailure(throwable, detailsConverter); + } + }; + converter = + TemporalTransferTypeDataConverter.wrap( + DefaultDataConverter.newDefaultInstance().withFailureConverter(failureConverter)); + + Failure failure = + converter.exceptionToFailure( + ApplicationFailure.newFailure("message", "type", new Model("detail"))); + ApplicationFailure restored = (ApplicationFailure) converter.failureToException(failure); + + assertEquals(new Model("detail"), restored.getDetails().get(0, Model.class, Model.class)); + } + + @Test + public void wrapperIsIdempotentAndReusesConverterInstances() { + assertSame(converter, TemporalTransferTypeDataConverter.wrap(converter)); + + converter.toPayload(new ReuseModel("one")); + converter.toPayload(new ReuseModel("two")); + assertEquals(1, ReuseModelConverter.instances.get()); + } + + @Test + public void nullRawAndUnannotatedValuesPassThrough() { + assertEquals( + DefaultDataConverter.STANDARD_PAYLOAD_CONVERTERS[0].getEncodingType(), + converter.toPayload(null).get().getMetadataOrThrow("encoding").toStringUtf8()); + + Payload rawPayload = DefaultDataConverter.newDefaultInstance().toPayload("raw").get(); + assertEquals(rawPayload, converter.toPayload(new RawValue(rawPayload)).get()); + assertEquals( + rawPayload, converter.fromPayload(rawPayload, RawValue.class, RawValue.class).getPayload()); + assertEquals( + "plain", + converter.fromPayload(converter.toPayload("plain").get(), String.class, String.class)); + } + + @Test + public void nullTransferValueIsReconstructed() { + Payload payload = converter.toPayload(new NullRepresentationModel()).get(); + + NullRepresentationModel restored = + converter.fromPayload( + payload, NullRepresentationModel.class, NullRepresentationModel.class); + + assertEquals( + DefaultDataConverter.STANDARD_PAYLOAD_CONVERTERS[0].getEncodingType(), + payload.getMetadataOrThrow("encoding").toStringUtf8()); + assertTrue(restored.reconstructed); + } + + @Test + public void mixedBatchesPreserveOrderAndMissingValues() { + Optional payloads = converter.toPayloads(new Model("one"), "two", new Model("three")); + Object[] values = + converter.fromPayloads( + payloads, + new Class[] {Model.class, String.class, Model.class, Model.class}, + new Type[] {Model.class, String.class, Model.class, Model.class}); + + assertEquals(new Model("one"), values[0]); + assertEquals("two", values[1]); + assertEquals(new Model("three"), values[2]); + assertNull(values[3]); + assertNull(converter.fromPayloads(4, payloads, Model.class, Model.class)); + } + + @Test + public void annotationIsExactAndDerivedClassMayOwnItsConverter() { + Payload inherited = converter.toPayload(new DerivedWithoutAnnotation("value")).get(); + assertEquals("json/plain", inherited.getMetadataOrThrow("encoding").toStringUtf8()); + + Payload own = converter.toPayload(new DerivedWithAnnotation("value")).get(); + assertEquals("json/protobuf", own.getMetadataOrThrow("encoding").toStringUtf8()); + assertEquals( + new DerivedWithAnnotation("value"), + converter.fromPayload(own, DerivedWithAnnotation.class, DerivedWithAnnotation.class)); + } + + @Test + public void inboundLookupDoesNotInheritBaseConverter() { + DataConverter delegate = mock(DataConverter.class); + DataConverter transferAware = TemporalTransferTypeDataConverter.wrap(delegate); + Payload payload = Payload.getDefaultInstance(); + DerivedWithoutAnnotation expected = new DerivedWithoutAnnotation("value"); + when(delegate.fromPayload( + payload, DerivedWithoutAnnotation.class, DerivedWithoutAnnotation.class)) + .thenReturn(expected); + + DerivedWithoutAnnotation actual = + transferAware.fromPayload( + payload, DerivedWithoutAnnotation.class, DerivedWithoutAnnotation.class); + + assertSame(expected, actual); + verify(delegate) + .fromPayload(payload, DerivedWithoutAnnotation.class, DerivedWithoutAnnotation.class); + } + + @Test + public void conversionIsTopLevelAndPerformsOneTransferStep() { + SecondStepConverter.invocations.set(0); + NestedModelConverter.invocations.set(0); + + FirstStepModel firstStep = + converter.fromPayload( + converter.toPayload(new FirstStepModel("one")).get(), + FirstStepModel.class, + FirstStepModel.class); + OrdinaryContainer container = + converter.fromPayload( + converter.toPayload(new OrdinaryContainer(new NestedModel("two"))).get(), + OrdinaryContainer.class, + OrdinaryContainer.class); + + assertEquals("one", firstStep.value); + assertEquals(0, SecondStepConverter.invocations.get()); + assertEquals("two", container.value.value); + assertEquals(0, NestedModelConverter.invocations.get()); + } + + @Test + public void genericRequestedTypeIsPreservedAndSelectsTransferType() { + Type stringType = new TypeToken>() {}.getType(); + Type integerType = new TypeToken>() {}.getType(); + + Payload stringPayload = converter.toPayload(new GenericValue("text")).get(); + Payload integerPayload = converter.toPayload(new GenericValue(12)).get(); + GenericValue stringValue = + converter.fromPayload(stringPayload, GenericValue.class, stringType); + GenericValue integerValue = + converter.fromPayload(integerPayload, GenericValue.class, integerType); + + assertEquals("text", stringValue.value); + assertEquals(Integer.valueOf(12), integerValue.value); + assertSame(integerType, GenericValueConverter.lastRequestedType); + } + + @Test + public void classIsUsedWhenRequestedTypeIsAbsent() { + Payload payload = converter.toPayload(new Model("value")).get(); + + assertEquals(new Model("value"), converter.fromPayload(payload, Model.class, null)); + assertSame(Model.class, ModelConverter.lastTransferTypeRequest); + assertSame(Model.class, ModelConverter.lastConversionRequest); + + Optional payloads = converter.toPayloads(new Model("value")); + assertEquals(new Model("value"), converter.fromPayloads(0, payloads, Model.class, null)); + assertSame(Model.class, ModelConverter.lastTransferTypeRequest); + assertSame(Model.class, ModelConverter.lastConversionRequest); + + Object[] values = + converter.fromPayloads(payloads, new Class[] {Model.class}, new Type[] {null}); + assertEquals(new Model("value"), values[0]); + assertSame(Model.class, ModelConverter.lastTransferTypeRequest); + assertSame(Model.class, ModelConverter.lastConversionRequest); + } + + @Test + public void parameterizedTransferTypeIsPassedIntact() { + Type modelType = new TypeToken>() {}.getType(); + ListModel value = + converter.fromPayload( + converter.toPayload(new ListModel(Arrays.asList("one", "two"))).get(), + ListModel.class, + modelType); + + assertEquals("one", value.values.get(0)); + assertEquals("two", value.values.get(1)); + assertTrue(ListModelConverter.lastTransferType instanceof ParameterizedType); + } + + @Test + public void invalidDeclarationsFailAsDataConverterExceptions() { + assertInvalid(AbstractModel.class); + assertInvalid(MissingPublicConstructorModel.class); + assertInvalid(ThrowingConstructorModel.class); + + Payload payload = DefaultDataConverter.newDefaultInstance().toPayload("value").get(); + assertThrows( + DataConverterException.class, + () -> converter.fromPayload(payload, NullTransferModel.class, NullTransferModel.class)); + } + + @Test + public void mismatchedConverterResultFailsAtTheArrayDecodeBoundary() { + Payload payload = DefaultDataConverter.newDefaultInstance().toPayload("value").get(); + Optional payloads = Optional.of(Payloads.newBuilder().addPayloads(payload).build()); + + assertMismatchedResult( + assertThrows( + DataConverterException.class, + () -> converter.fromPayload(payload, MismatchedModel.class, MismatchedModel.class))); + assertMismatchedResult( + assertThrows( + DataConverterException.class, + () -> + converter.fromPayloads(0, payloads, MismatchedModel.class, MismatchedModel.class))); + assertMismatchedResult( + assertThrows( + DataConverterException.class, + () -> + converter.fromPayloads( + payloads, + new Class[] {MismatchedModel.class}, + new Type[] {MismatchedModel.class}))); + } + + @Test + public void callbackFailuresPropagateUnchanged() { + CallbackException expected = + assertThrows(CallbackException.class, () -> converter.toPayload(new FailingModel())); + assertEquals("callback", expected.getMessage()); + } + + @Test + public void configuredConverterRemainsAuthoritativeAndReceivesContext() { + DataConverter delegate = mock(DataConverter.class); + DataConverter contextualDelegate = mock(DataConverter.class); + SerializationContext context = mock(SerializationContext.class); + Payload payload = Payload.getDefaultInstance(); + when(delegate.withContext(context)).thenReturn(contextualDelegate); + when(contextualDelegate.toPayload(any())).thenReturn(Optional.of(payload)); + + DataConverter contextual = + TemporalTransferTypeDataConverter.wrap(delegate).withContext(context); + assertSame(payload, contextual.toPayload(new Model("value")).get()); + + ArgumentCaptor transferred = ArgumentCaptor.forClass(Object.class); + verify(contextualDelegate).toPayload(transferred.capture()); + assertEquals(StringValue.of("value"), transferred.getValue()); + verify(delegate, never()).toPayload(any()); + } + + private void assertInvalid(Class modelClass) { + DataConverterException exception = + assertThrows( + DataConverterException.class, + () -> converter.toPayload(modelClass.getDeclaredConstructor().newInstance())); + assertTrue(exception.getMessage().contains(modelClass.getName())); + } + + private void assertMismatchedResult(DataConverterException exception) { + assertTrue(exception.getMessage().contains(MismatchedConverter.class.getName())); + assertTrue(exception.getMessage().contains(MismatchedModel.class.getName())); + assertTrue(exception.getMessage().contains(String.class.getName())); + } + + @TransferTypeConvertible(ModelConverter.class) + public static class Model { + public final String value; + + public Model(String value) { + this.value = value; + } + + @Override + public boolean equals(Object other) { + return other instanceof Model && value.equals(((Model) other).value); + } + + @Override + public int hashCode() { + return value.hashCode(); + } + } + + public static final class ModelConverter implements TransferTypeConverter { + static Type lastTransferTypeRequest; + static Type lastConversionRequest; + + public ModelConverter() {} + + @Override + public Type getTransferType(Type valueType) { + lastTransferTypeRequest = valueType; + return StringValue.class; + } + + @Override + public Object toTransferType(Model value) { + return StringValue.of(value.value); + } + + @Override + public Model fromTransferType(Object value, Type valueType) { + lastConversionRequest = valueType; + return new Model(((StringValue) value).getValue()); + } + } + + @TransferTypeConvertible(ReuseModelConverter.class) + public static final class ReuseModel extends Model { + public ReuseModel(String value) { + super(value); + } + } + + public static final class ReuseModelConverter implements TransferTypeConverter { + static final AtomicInteger instances = new AtomicInteger(); + + public ReuseModelConverter() { + instances.incrementAndGet(); + } + + @Override + public Type getTransferType(Type valueType) { + return StringValue.class; + } + + @Override + public Object toTransferType(ReuseModel value) { + return StringValue.of(value.value); + } + + @Override + public ReuseModel fromTransferType(Object value, Type valueType) { + return new ReuseModel(((StringValue) value).getValue()); + } + } + + public static class DerivedWithoutAnnotation extends Model { + public DerivedWithoutAnnotation(String value) { + super(value); + } + } + + @TransferTypeConvertible(DerivedConverter.class) + public static class DerivedWithAnnotation extends Model { + public DerivedWithAnnotation(String value) { + super(value); + } + } + + public static final class DerivedConverter + implements TransferTypeConverter { + public DerivedConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return StringValue.class; + } + + @Override + public Object toTransferType(DerivedWithAnnotation value) { + return StringValue.of(value.value); + } + + @Override + public DerivedWithAnnotation fromTransferType(Object value, Type valueType) { + return new DerivedWithAnnotation(((StringValue) value).getValue()); + } + } + + @TransferTypeConvertible(GenericValueConverter.class) + public static final class GenericValue { + final T value; + + GenericValue(T value) { + this.value = value; + } + } + + public static final class GenericValueConverter + implements TransferTypeConverter> { + static Type lastRequestedType; + + public GenericValueConverter() {} + + @Override + public Type getTransferType(Type valueType) { + lastRequestedType = valueType; + return ((ParameterizedType) valueType).getActualTypeArguments()[0]; + } + + @Override + public Object toTransferType(GenericValue value) { + return value.value; + } + + @Override + public GenericValue fromTransferType(Object value, Type valueType) { + lastRequestedType = valueType; + return new GenericValue(value); + } + } + + @TransferTypeConvertible(ListModelConverter.class) + public static final class ListModel { + final List values; + + ListModel(List values) { + this.values = values; + } + } + + public static final class ListModelConverter implements TransferTypeConverter> { + static Type lastTransferType; + + public ListModelConverter() {} + + @Override + public Type getTransferType(Type valueType) { + lastTransferType = new TypeToken>() {}.getType(); + return lastTransferType; + } + + @Override + public Object toTransferType(ListModel value) { + return value.values; + } + + @Override + public ListModel fromTransferType(Object value, Type valueType) { + return new ListModel((List) value); + } + } + + @TransferTypeConvertible(AbstractConverter.class) + public static final class AbstractModel { + public AbstractModel() {} + } + + public abstract static class AbstractConverter implements TransferTypeConverter {} + + @TransferTypeConvertible(MissingPublicConstructorConverter.class) + public static final class MissingPublicConstructorModel { + public MissingPublicConstructorModel() {} + } + + public static final class MissingPublicConstructorConverter + implements TransferTypeConverter { + private MissingPublicConstructorConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return String.class; + } + + @Override + public Object toTransferType(MissingPublicConstructorModel value) { + return "value"; + } + + @Override + public MissingPublicConstructorModel fromTransferType(Object value, Type valueType) { + return new MissingPublicConstructorModel(); + } + } + + @TransferTypeConvertible(ThrowingConstructorConverter.class) + public static final class ThrowingConstructorModel { + public ThrowingConstructorModel() {} + } + + public static final class ThrowingConstructorConverter + implements TransferTypeConverter { + public ThrowingConstructorConverter() { + throw new IllegalStateException("constructor"); + } + + @Override + public Type getTransferType(Type valueType) { + return String.class; + } + + @Override + public Object toTransferType(ThrowingConstructorModel value) { + return "value"; + } + + @Override + public ThrowingConstructorModel fromTransferType(Object value, Type valueType) { + return new ThrowingConstructorModel(); + } + } + + @TransferTypeConvertible(NullTransferConverter.class) + public static final class NullTransferModel {} + + public static final class NullTransferConverter + implements TransferTypeConverter { + public NullTransferConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return null; + } + + @Override + public Object toTransferType(NullTransferModel value) { + return "value"; + } + + @Override + public NullTransferModel fromTransferType(Object value, Type valueType) { + return new NullTransferModel(); + } + } + + @TransferTypeConvertible(MismatchedConverter.class) + public static final class MismatchedModel {} + + public static final class MismatchedConverter implements TransferTypeConverter { + public MismatchedConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return String.class; + } + + @Override + public Object toTransferType(String value) { + return value; + } + + @Override + public String fromTransferType(Object value, Type valueType) { + return (String) value; + } + } + + @TransferTypeConvertible(FailingConverter.class) + public static final class FailingModel {} + + public static final class FailingConverter implements TransferTypeConverter { + public FailingConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return String.class; + } + + @Override + public Object toTransferType(FailingModel value) { + throw new CallbackException("callback"); + } + + @Override + public FailingModel fromTransferType(Object value, Type valueType) { + throw new CallbackException("callback"); + } + } + + @TransferTypeConvertible(NullRepresentationConverter.class) + public static final class NullRepresentationModel { + private final boolean reconstructed; + + public NullRepresentationModel() { + this(false); + } + + private NullRepresentationModel(boolean reconstructed) { + this.reconstructed = reconstructed; + } + } + + public static final class NullRepresentationConverter + implements TransferTypeConverter { + public NullRepresentationConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return String.class; + } + + @Override + public Object toTransferType(NullRepresentationModel value) { + return null; + } + + @Override + public NullRepresentationModel fromTransferType(Object value, Type valueType) { + assertNull(value); + return new NullRepresentationModel(true); + } + } + + @TransferTypeConvertible(FirstStepConverter.class) + public static final class FirstStepModel { + private final String value; + + private FirstStepModel(String value) { + this.value = value; + } + } + + public static final class FirstStepConverter implements TransferTypeConverter { + public FirstStepConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return SecondStepModel.class; + } + + @Override + public Object toTransferType(FirstStepModel value) { + return new SecondStepModel(value.value); + } + + @Override + public FirstStepModel fromTransferType(Object value, Type valueType) { + return new FirstStepModel(((SecondStepModel) value).value); + } + } + + @TransferTypeConvertible(SecondStepConverter.class) + public static final class SecondStepModel { + public String value; + + public SecondStepModel() {} + + private SecondStepModel(String value) { + this.value = value; + } + } + + public static final class SecondStepConverter implements TransferTypeConverter { + private static final AtomicInteger invocations = new AtomicInteger(); + + public SecondStepConverter() {} + + @Override + public Type getTransferType(Type valueType) { + invocations.incrementAndGet(); + return String.class; + } + + @Override + public Object toTransferType(SecondStepModel value) { + invocations.incrementAndGet(); + return value.value; + } + + @Override + public SecondStepModel fromTransferType(Object value, Type valueType) { + invocations.incrementAndGet(); + return new SecondStepModel((String) value); + } + } + + public static final class OrdinaryContainer { + public NestedModel value; + + public OrdinaryContainer() {} + + private OrdinaryContainer(NestedModel value) { + this.value = value; + } + } + + @TransferTypeConvertible(NestedModelConverter.class) + public static final class NestedModel { + public String value; + + public NestedModel() {} + + private NestedModel(String value) { + this.value = value; + } + } + + public static final class NestedModelConverter implements TransferTypeConverter { + private static final AtomicInteger invocations = new AtomicInteger(); + + public NestedModelConverter() {} + + @Override + public Type getTransferType(Type valueType) { + invocations.incrementAndGet(); + return String.class; + } + + @Override + public Object toTransferType(NestedModel value) { + invocations.incrementAndGet(); + return value.value; + } + + @Override + public NestedModel fromTransferType(Object value, Type valueType) { + invocations.incrementAndGet(); + return new NestedModel((String) value); + } + } + + private static final class CallbackException extends RuntimeException { + private CallbackException(String message) { + super(message); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerImplTest.java index 2ddfa7fda1..b37bf1b377 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/nexus/NexusTaskHandlerImplTest.java @@ -1,6 +1,7 @@ package io.temporal.internal.nexus; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import com.google.protobuf.ByteString; import com.uber.m3.tally.RootScopeBuilder; @@ -17,6 +18,7 @@ import io.temporal.api.nexus.v1.StartOperationResponse; import io.temporal.api.workflowservice.v1.PollNexusTaskQueueResponse; import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DefaultDataConverter; import io.temporal.common.interceptors.WorkerInterceptor; @@ -105,6 +107,36 @@ public void startSyncTask() throws TimeoutException { String.class)); } + @Test + public void nexusOperationContextExposesConfiguredConverter() throws TimeoutException { + DataConverter userDataConverter = DefaultDataConverter.newDefaultInstance(); + WorkflowClient client = mock(WorkflowClient.class); + when(client.getOptions()) + .thenReturn(WorkflowClientOptions.newBuilder().setDataConverter(userDataConverter).build()); + ConverterContextNexusService service = new ConverterContextNexusService(userDataConverter); + NexusTaskHandlerImpl nexusTaskHandlerImpl = + new NexusTaskHandlerImpl( + client, NAMESPACE, TASK_QUEUE, dataConverter, new WorkerInterceptor[] {}); + nexusTaskHandlerImpl.registerNexusServiceImplementations(new Object[] {service}); + nexusTaskHandlerImpl.start(); + + PollNexusTaskQueueResponse.Builder task = + PollNexusTaskQueueResponse.newBuilder() + .setRequest( + Request.newBuilder() + .setStartOperation( + StartOperationRequest.newBuilder() + .setOperation("operation") + .setService("TestNexusService1") + .setPayload(dataConverter.toPayload("world").get()) + .build())); + + NexusTaskHandler.Result result = + nexusTaskHandlerImpl.handle(new NexusTask(task, null, null), metricsScope); + Assert.assertNull(result.getHandlerException()); + Assert.assertTrue(service.usesConfiguredConverter); + } + @Test public void syncTimeoutTask() { WorkflowClient client = mock(WorkflowClient.class); @@ -414,6 +446,30 @@ public OperationHandler operation() { } } + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public class ConverterContextNexusService { + private final DataConverter expectedConverter; + private boolean usesConfiguredConverter; + + ConverterContextNexusService(DataConverter expectedConverter) { + this.expectedConverter = expectedConverter; + } + + @OperationImpl + public OperationHandler operation() { + return OperationHandler.sync( + (ctx, details, name) -> { + usesConfiguredConverter = + io.temporal.nexus.Nexus.getOperationContext() + .getWorkflowClient() + .getOptions() + .getDataConverter() + == expectedConverter; + return name; + }); + } + } + @ServiceImpl(service = TestNexusServices.TestNexusService2.class) public class TestNexusServiceImpl2 { @OperationImpl diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java index b7ffce91ca..bb2a4606ef 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java @@ -43,15 +43,17 @@ private Worker buildWorker(WorkerOptions options) { when(service.blockingStub()).thenReturn(blockingStub); when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + WorkflowClientOptions clientOptions = + WorkflowClientOptions.newBuilder() + .setNamespace("test-ns") + .setIdentity("test-worker") + .validateAndBuildWithDefaults(); + WorkflowClientInternal clientInternal = mock(WorkflowClientInternal.class); + when(clientInternal.getInternalDataConverter()).thenReturn(clientOptions.getDataConverter()); WorkflowClient client = mock(WorkflowClient.class); - when(client.getInternal()).thenReturn(mock(WorkflowClientInternal.class)); + when(client.getInternal()).thenReturn(clientInternal); when(client.getWorkflowServiceStubs()).thenReturn(service); - when(client.getOptions()) - .thenReturn( - WorkflowClientOptions.newBuilder() - .setNamespace("test-ns") - .setIdentity("test-worker") - .validateAndBuildWithDefaults()); + when(client.getOptions()).thenReturn(clientOptions); Scope metricsScope = new NoopScope(); WorkflowRunLockManager runLocks = new WorkflowRunLockManager(); diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java index 1d5f5df30d..261c4f47a4 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java @@ -97,15 +97,17 @@ public void autoEnrollAtStartupSwitchesPollersToAutoscaling() throws Exception { when(service.blockingStub()).thenReturn(blockingStub); when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + WorkflowClientOptions clientOptions = + WorkflowClientOptions.newBuilder() + .setNamespace("test-ns") + .setIdentity("test-worker") + .validateAndBuildWithDefaults(); + WorkflowClientInternal clientInternal = mock(WorkflowClientInternal.class); + when(clientInternal.getInternalDataConverter()).thenReturn(clientOptions.getDataConverter()); WorkflowClient client = mock(WorkflowClient.class); - when(client.getInternal()).thenReturn(mock(WorkflowClientInternal.class)); + when(client.getInternal()).thenReturn(clientInternal); when(client.getWorkflowServiceStubs()).thenReturn(service); - when(client.getOptions()) - .thenReturn( - WorkflowClientOptions.newBuilder() - .setNamespace("test-ns") - .setIdentity("test-worker") - .validateAndBuildWithDefaults()); + when(client.getOptions()).thenReturn(clientOptions); // Namespace advertises the auto-enroll capability. NamespaceCapabilities capabilities = new NamespaceCapabilities(); diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java index 07af0f5bec..5345f232b1 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java @@ -205,15 +205,17 @@ private static Worker newWorker(WorkflowServiceGrpc.WorkflowServiceFutureStub fu when(service.blockingStub()).thenReturn(blockingStub); when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + WorkflowClientOptions clientOptions = + WorkflowClientOptions.newBuilder() + .setNamespace("test-ns") + .setIdentity("test-worker") + .validateAndBuildWithDefaults(); + WorkflowClientInternal clientInternal = mock(WorkflowClientInternal.class); + when(clientInternal.getInternalDataConverter()).thenReturn(clientOptions.getDataConverter()); WorkflowClient client = mock(WorkflowClient.class); - when(client.getInternal()).thenReturn(mock(WorkflowClientInternal.class)); + when(client.getInternal()).thenReturn(clientInternal); when(client.getWorkflowServiceStubs()).thenReturn(service); - when(client.getOptions()) - .thenReturn( - WorkflowClientOptions.newBuilder() - .setNamespace("test-ns") - .setIdentity("test-worker") - .validateAndBuildWithDefaults()); + when(client.getOptions()).thenReturn(clientOptions); Scope metricsScope = new NoopScope(); WorkflowRunLockManager runLocks = new WorkflowRunLockManager(); diff --git a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java index dc2a0d3d3c..695b16f034 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java +++ b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java @@ -78,6 +78,7 @@ public final class TestActivityEnvironmentInternal implements TestActivityEnviro private final InProcessGRPCServer mockServer; private final ActivityTaskHandlerImpl activityTaskHandler; private final TestEnvironmentOptions testEnvironmentOptions; + private final DataConverter dataConverter; private final WorkflowServiceStubs workflowServiceStubs; private final AtomicReference heartbeatDetails = new AtomicReference<>(); private final DataConverter heartbeatDetailsConverter; @@ -104,35 +105,33 @@ public TestActivityEnvironmentInternal(@Nullable TestEnvironmentOptions options) this.workflowServiceStubs = WorkflowServiceStubs.newServiceStubs(serviceStubsOptionsBuilder.build()); - WorkflowClient client = + WorkflowClient workflowClient = WorkflowClient.newInstance( this.workflowServiceStubs, testEnvironmentOptions.getWorkflowClientOptions()); + this.dataConverter = + ((WorkflowClientInternal) workflowClient.getInternal()).getInternalDataConverter(); ExternalStorageRunner externalStorageRunner = - ((WorkflowClientInternal) client.getInternal()).getExternalStorageRunner(); - DataConverter clientDataConverter = - testEnvironmentOptions.getWorkflowClientOptions().getDataConverter(); + ((WorkflowClientInternal) workflowClient.getInternal()).getExternalStorageRunner(); this.heartbeatDetailsConverter = - externalStorageRunner == null - ? clientDataConverter - : new ExternalStorageDataConverter(clientDataConverter, externalStorageRunner); + new ExternalStorageDataConverter(dataConverter, externalStorageRunner); ActivityExecutionContextFactory activityExecutionContextFactory = new ActivityExecutionContextFactoryImpl( - client, - testEnvironmentOptions.getWorkflowClientOptions().getIdentity(), - testEnvironmentOptions.getWorkflowClientOptions().getNamespace(), + workflowClient, + workflowClient.getOptions().getIdentity(), + workflowClient.getOptions().getNamespace(), WorkerOptions.getDefaultInstance().getMaxHeartbeatThrottleInterval(), WorkerOptions.getDefaultInstance().getDefaultHeartbeatThrottleInterval(), - clientDataConverter, + dataConverter, heartbeatExecutor, externalStorageRunner); activityTaskHandler = new ActivityTaskHandlerImpl( - testEnvironmentOptions.getWorkflowClientOptions().getNamespace(), + workflowClient.getOptions().getNamespace(), "test-activity-env-task-queue", - testEnvironmentOptions.getWorkflowClientOptions().getDataConverter(), + dataConverter, activityExecutionContextFactory, testEnvironmentOptions.getWorkerFactoryOptions().getWorkerInterceptors(), - testEnvironmentOptions.getWorkflowClientOptions().getContextPropagators()); + workflowClient.getOptions().getContextPropagators()); } private class HeartbeatInterceptingService extends WorkflowServiceGrpc.WorkflowServiceImplBase { @@ -272,19 +271,10 @@ private class TestActivityExecutor implements WorkflowOutboundCallsInterceptor { @Override public ActivityOutput executeActivity(ActivityInput i) { - Optional payloads = - testEnvironmentOptions - .getWorkflowClientOptions() - .getDataConverter() - .toPayloads(i.getArgs()); + Optional payloads = dataConverter.toPayloads(i.getArgs()); Optional heartbeatPayload = Optional.ofNullable(heartbeatDetails.getAndSet(null)) - .flatMap( - obj -> - testEnvironmentOptions - .getWorkflowClientOptions() - .getDataConverter() - .toPayloads(obj)); + .flatMap(obj -> dataConverter.toPayloads(obj)); ActivityOptions options = i.getOptions(); PollActivityTaskQueueResponse.Builder taskBuilder = @@ -315,11 +305,7 @@ public ActivityOutput executeActivity(ActivityInput i) { @Override public LocalActivityOutput executeLocalActivity(LocalActivityInput i) { - Optional payloads = - testEnvironmentOptions - .getWorkflowClientOptions() - .getDataConverter() - .toPayloads(i.getArgs()); + Optional payloads = dataConverter.toPayloads(i.getArgs()); LocalActivityOptions options = i.getOptions(); PollActivityTaskQueueResponse.Builder taskBuilder = PollActivityTaskQueueResponse.newBuilder() @@ -523,8 +509,6 @@ private T getReply( ActivityTaskHandler.Result response, Class resultClass, Type resultType) { - DataConverter dataConverter = - testEnvironmentOptions.getWorkflowClientOptions().getDataConverter(); if (response.getTaskCompleted() != null) { RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted(); Optional result = diff --git a/temporal-testing/src/main/java/io/temporal/testing/internal/SDKTestWorkflowRule.java b/temporal-testing/src/main/java/io/temporal/testing/internal/SDKTestWorkflowRule.java index 9c81461212..37e2b914a7 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/internal/SDKTestWorkflowRule.java +++ b/temporal-testing/src/main/java/io/temporal/testing/internal/SDKTestWorkflowRule.java @@ -270,14 +270,16 @@ public Endpoint getNexusEndpoint() { } /** - * Returns a {@link NexusClient} bound to this rule's namespace and service stubs. Use for tests - * that exercise the standalone Nexus client surface. + * Returns a {@link NexusClient} bound to this rule's namespace, configured data converter, and + * service stubs. Use for tests that exercise the standalone Nexus client surface. */ public NexusClient getNexusClient() { + WorkflowClientOptions options = getWorkflowClient().getOptions(); return NexusClient.newInstance( getWorkflowServiceStubs(), NexusClientOptions.newBuilder() - .setNamespace(getWorkflowClient().getOptions().getNamespace()) + .setNamespace(options.getNamespace()) + .setDataConverter(options.getDataConverter()) .build()); } diff --git a/temporal-testing/src/test/java/io/temporal/testing/TestActivityEnvironmentTransferTypeTest.java b/temporal-testing/src/test/java/io/temporal/testing/TestActivityEnvironmentTransferTypeTest.java new file mode 100644 index 0000000000..d053524cfd --- /dev/null +++ b/temporal-testing/src/test/java/io/temporal/testing/TestActivityEnvironmentTransferTypeTest.java @@ -0,0 +1,284 @@ +package io.temporal.testing; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.protobuf.StringValue; +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.LocalActivityOptions; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.SimplePlugin; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterException; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.TransferTypeConverter; +import io.temporal.common.converter.TransferTypeConvertible; +import io.temporal.payload.context.SerializationContext; +import java.lang.reflect.Type; +import java.time.Duration; +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; +import javax.annotation.Nonnull; +import org.junit.jupiter.api.Test; + +class TestActivityEnvironmentTransferTypeTest { + + @Test + void activityContextExposesTransferAwareConverter() { + TrackingDataConverter trackingConverter = new TrackingDataConverter(); + TestActivityEnvironment environment = newEnvironment(trackingConverter); + try { + environment.registerActivitiesImplementations(new ConverterActivityImpl(trackingConverter)); + ConverterActivity activity = environment.newActivityStub(ConverterActivity.class); + + assertTrue(activity.usesTransferAwareConverter()); + } finally { + environment.close(); + } + } + + @Test + void wrapsConfiguredConverterForArgumentsAndResults() { + TrackingDataConverter trackingConverter = new TrackingDataConverter(); + TestActivityEnvironment environment = newEnvironment(trackingConverter); + try { + environment.registerActivitiesImplementations(new TransferActivityImpl()); + TransferActivity activity = environment.newActivityStub(TransferActivity.class); + + TransferModel result = activity.execute(new TransferModel("value")); + assertEquals(new TransferModel("value-result"), result); + assertTrue(result.wasTransferred()); + assertTrue(trackingConverter.toPayloadCalls >= 2); + assertTrue(trackingConverter.fromPayloadCalls >= 2); + } finally { + environment.close(); + } + } + + @Test + void wrapsConfiguredConverterForLocalActivityArgumentsAndResults() { + TrackingDataConverter trackingConverter = new TrackingDataConverter(); + TestActivityEnvironment environment = newEnvironment(trackingConverter); + try { + environment.registerActivitiesImplementations(new TransferActivityImpl()); + TransferActivity activity = + environment.newLocalActivityStub( + TransferActivity.class, + LocalActivityOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofMinutes(1)) + .build(), + Collections.emptyMap()); + + TransferModel result = activity.execute(new TransferModel("value")); + assertEquals(new TransferModel("value-result"), result); + assertTrue(result.wasTransferred()); + assertTrue(trackingConverter.toPayloadCalls >= 2); + assertTrue(trackingConverter.fromPayloadCalls >= 2); + } finally { + environment.close(); + } + } + + @Test + void wrapsConfiguredConverterForHeartbeatDetails() { + TrackingDataConverter trackingConverter = new TrackingDataConverter(); + TestActivityEnvironment environment = newEnvironment(trackingConverter); + try { + environment.registerActivitiesImplementations(new HeartbeatActivityImpl()); + AtomicReference heartbeat = new AtomicReference<>(); + environment.setHeartbeatDetails(new TransferModel("initial")); + environment.setActivityHeartbeatListener(TransferModel.class, heartbeat::set); + HeartbeatActivity activity = environment.newActivityStub(HeartbeatActivity.class); + + TransferModel result = activity.execute(); + assertEquals(new TransferModel("initial"), result); + assertTrue(result.wasTransferred()); + assertEquals(new TransferModel("initial-heartbeat"), heartbeat.get()); + assertTrue(heartbeat.get().wasTransferred()); + assertTrue(trackingConverter.toPayloadCalls >= 3); + assertTrue(trackingConverter.fromPayloadCalls >= 3); + } finally { + environment.close(); + } + } + + private TestActivityEnvironment newEnvironment(TrackingDataConverter trackingConverter) { + DataConverter originalConverter = DefaultDataConverter.newDefaultInstance(); + TestEnvironmentOptions options = + TestEnvironmentOptions.newBuilder() + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setDataConverter(originalConverter) + .setPlugins(new ConverterPlugin(trackingConverter)) + .build()) + .build(); + assertEquals(originalConverter, options.getWorkflowClientOptions().getDataConverter()); + return TestActivityEnvironment.newInstance(options); + } + + private static final class ConverterPlugin extends SimplePlugin { + private final DataConverter dataConverter; + + private ConverterPlugin(DataConverter dataConverter) { + super("test-activity-environment-converter"); + this.dataConverter = dataConverter; + } + + @Override + public void configureWorkflowClient(@Nonnull WorkflowClientOptions.Builder builder) { + builder.setDataConverter(dataConverter); + } + } + + private static final class TrackingDataConverter implements DataConverter { + private final DataConverter delegate = DefaultDataConverter.newDefaultInstance(); + private int toPayloadCalls; + private int fromPayloadCalls; + + @Override + public Optional toPayload(T value) throws DataConverterException { + toPayloadCalls++; + return delegate.toPayload(value); + } + + @Override + public T fromPayload(Payload payload, Class valueClass, Type valueType) + throws DataConverterException { + fromPayloadCalls++; + return delegate.fromPayload(payload, valueClass, valueType); + } + + @Override + public Optional toPayloads(Object... values) throws DataConverterException { + toPayloadCalls++; + return delegate.toPayloads(values); + } + + @Override + public T fromPayloads( + int index, Optional content, Class valueClass, Type valueType) + throws DataConverterException { + fromPayloadCalls++; + return delegate.fromPayloads(index, content, valueClass, valueType); + } + + @Override + public DataConverter withContext(@Nonnull SerializationContext context) { + return this; + } + } + + @ActivityInterface + public interface TransferActivity { + @ActivityMethod + TransferModel execute(TransferModel input); + } + + public static final class TransferActivityImpl implements TransferActivity { + @Override + public TransferModel execute(TransferModel input) { + if (!input.wasTransferred()) { + throw new IllegalStateException("Activity input did not use its transfer type converter"); + } + return new TransferModel(input.value + "-result"); + } + } + + @ActivityInterface + public interface ConverterActivity { + @ActivityMethod + boolean usesTransferAwareConverter(); + } + + private static final class ConverterActivityImpl implements ConverterActivity { + private final TrackingDataConverter configuredConverter; + + private ConverterActivityImpl(TrackingDataConverter configuredConverter) { + this.configuredConverter = configuredConverter; + } + + @Override + public boolean usesTransferAwareConverter() { + DataConverter exposed = + Activity.getExecutionContext().getWorkflowClient().getOptions().getDataConverter(); + int callsBefore = configuredConverter.toPayloadCalls; + Payload payload = exposed.toPayload(new TransferModel("value")).get(); + return exposed != configuredConverter + && configuredConverter.toPayloadCalls > callsBefore + && "json/protobuf".equals(payload.getMetadataOrThrow("encoding").toStringUtf8()); + } + } + + @ActivityInterface + public interface HeartbeatActivity { + @ActivityMethod + TransferModel execute(); + } + + public static final class HeartbeatActivityImpl implements HeartbeatActivity { + @Override + public TransferModel execute() { + Optional details = + Activity.getExecutionContext().getHeartbeatDetails(TransferModel.class); + TransferModel value = details.orElse(null); + if (!value.wasTransferred()) { + throw new IllegalStateException("Heartbeat detail did not use its transfer type converter"); + } + Activity.getExecutionContext().heartbeat(new TransferModel(value.value + "-heartbeat")); + return value; + } + } + + @TransferTypeConvertible(TransferModelConverter.class) + public static final class TransferModel { + private final String value; + private final boolean transferred; + + private TransferModel(String value) { + this(value, false); + } + + private TransferModel(String value, boolean transferred) { + this.value = value; + this.transferred = transferred; + } + + private boolean wasTransferred() { + return transferred; + } + + @Override + public boolean equals(Object other) { + return other instanceof TransferModel && value.equals(((TransferModel) other).value); + } + + @Override + public int hashCode() { + return value.hashCode(); + } + } + + public static final class TransferModelConverter implements TransferTypeConverter { + public TransferModelConverter() {} + + @Override + public Type getTransferType(Type valueType) { + return StringValue.class; + } + + @Override + public Object toTransferType(TransferModel value) { + return StringValue.of(value.value); + } + + @Override + public TransferModel fromTransferType(Object value, Type valueType) { + return new TransferModel(((StringValue) value).getValue(), true); + } + } +} diff --git a/temporal-testing/src/test/java/io/temporal/testing/internal/SDKTestWorkflowRuleNexusClientTest.java b/temporal-testing/src/test/java/io/temporal/testing/internal/SDKTestWorkflowRuleNexusClientTest.java new file mode 100644 index 0000000000..1e077ae052 --- /dev/null +++ b/temporal-testing/src/test/java/io/temporal/testing/internal/SDKTestWorkflowRuleNexusClientTest.java @@ -0,0 +1,64 @@ +package io.temporal.testing.internal; + +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.temporal.api.common.v1.Payload; +import io.temporal.client.StartNexusOperationOptions; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.payload.context.SerializationContext; +import java.util.Optional; +import org.junit.jupiter.api.Test; + +public class SDKTestWorkflowRuleNexusClientTest { + private final RuntimeException expected = + new IllegalStateException("configured converter was used"); + private final DataConverter configuredConverter = new RejectingConverter(); + + @Test + public void nexusClientUsesConfiguredWorkflowClientConverter() { + SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder().setDataConverter(configuredConverter).build()) + .build(); + try { + RuntimeException thrown = + assertThrows( + RuntimeException.class, + () -> + testWorkflowRule + .getNexusClient() + .newUntypedNexusServiceClient("endpoint", "service") + .start( + "operation", + StartNexusOperationOptions.newBuilder().setId("id").build(), + "probe")); + + assertSame(expected, thrown); + } finally { + testWorkflowRule.getTestEnvironment().close(); + } + } + + private final class RejectingConverter extends DefaultDataConverter { + private RejectingConverter() { + super(STANDARD_PAYLOAD_CONVERTERS); + } + + @Override + public DataConverter withContext(SerializationContext context) { + return this; + } + + @Override + public Optional toPayload(T value) { + if ("probe".equals(value)) { + throw expected; + } + return super.toPayload(value); + } + } +}