diff --git a/contrib/temporal-workflowstreams/README.md b/contrib/temporal-workflowstreams/README.md index ad630c89b7..4e7dedf10b 100644 --- a/contrib/temporal-workflowstreams/README.md +++ b/contrib/temporal-workflowstreams/README.md @@ -124,9 +124,13 @@ automatically follows continue-as-new chains, recovers from truncation by restarting from the current base offset, and also ends when the owning `WorkflowStreamClient` is closed. -Items carry the raw `io.temporal.api.common.v1.Payload`; decode at the call -site with your data converter. Offsets are **global** (across all topics), not -per-topic. +Items carry the raw `io.temporal.api.common.v1.Payload`. Use +`WorkflowStreamClient.decodeItem` to decode them with the configured stream +item converter. Transfer conversion and payload conversion apply to each item. +Payload codecs apply only once to the surrounding Temporal signal or update +envelope, never to an individual item. Configure matching payload converters +on the workflow and client sides. Offsets are **global** (across all topics), +not per-topic. ### Listener (non-blocking) @@ -147,9 +151,7 @@ WorkflowStreamSubscriptionHandle handle = new WorkflowStreamListener() { @Override public CompletionStage onNext(WorkflowStreamItem item) { - String value = - DefaultDataConverter.STANDARD_INSTANCE.fromPayload( - item.getPayload(), String.class, String.class); + String value = client.decodeItem(item, String.class); System.out.printf( "offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value); return null; // or a pending stage to apply backpressure @@ -176,9 +178,7 @@ polling still runs on the shared executor: ```java try (WorkflowStreamSubscription subscription = client.subscribe(options)) { for (WorkflowStreamItem item : subscription) { - String value = - DefaultDataConverter.STANDARD_INSTANCE.fromPayload( - item.getPayload(), String.class, String.class); + String value = client.decodeItem(item, String.class); System.out.printf("offset=%d topic=%s value=%s%n", item.getOffset(), item.getTopic(), value); } } @@ -194,7 +194,7 @@ unrecoverable poll failure is rethrown from `hasNext()`. | `batchInterval` | 2s | Automatic flush interval | | `maxBatchSize` | unset | Flush once the buffer reaches this size | | `maxRetryDuration` | 10m | Max time to retry a failed flush before `FlushTimeoutException`. Must be < the workflow's publisher TTL (15m) to preserve exactly-once delivery | -| `payloadConverters` | standard set | Per-item serialization. Payload conversion only — the client's codec chain runs once on the envelope, never per item | +| `payloadConverters` | standard set | Per-item transfer and payload conversion. The client's codec chain runs once on the envelope, never per item | | `pollExecutor` | 2 daemon threads, client-owned | Scheduler shared by the client's subscriptions. It runs the short update-admission and delivery steps and poll cooldowns — never held during the long poll itself. A user-supplied executor is never shut down by the client; supply a bigger pool for many subscriptions against slow workflows | | `SubscribeOptions.pollCooldown` | 100ms | Min interval between polls | diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java index 32bfbc64bf..406f133b7f 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStream.java @@ -3,7 +3,6 @@ import io.temporal.api.common.v1.Payload; import io.temporal.common.Experimental; import io.temporal.common.converter.DataConverter; -import io.temporal.common.converter.DefaultDataConverter; import io.temporal.failure.ApplicationFailure; import io.temporal.workflow.ContinueAsNewOptions; import io.temporal.workflow.Workflow; @@ -77,13 +76,7 @@ public static WorkflowStream newInstance( } private WorkflowStream(@Nullable WorkflowStreamState priorState, WorkflowStreamOptions options) { - // A converter built only from PayloadConverters is codec-free, so workflow-published - // items are never double-encoded against the worker's response codec. - if (options.getPayloadConverters().length > 0) { - this.dataConverter = new DefaultDataConverter(options.getPayloadConverters()); - } else { - this.dataConverter = DefaultDataConverter.STANDARD_INSTANCE; - } + this.dataConverter = WorkflowStreamDataConverter.create(options.getPayloadConverters()); if (priorState != null) { baseOffset = priorState.baseOffset; diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java index b5348879d4..c3c3fa7ef1 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClient.java @@ -6,9 +6,9 @@ import io.temporal.client.WorkflowStub; import io.temporal.common.Experimental; import io.temporal.common.converter.DataConverter; -import io.temporal.common.converter.DefaultDataConverter; import io.temporal.workflowstreams.internal.StreamPublisher; import io.temporal.workflowstreams.internal.SubscriptionDriver; +import java.lang.reflect.Type; import java.util.HashMap; import java.util.Map; import java.util.Set; @@ -32,6 +32,7 @@ public final class WorkflowStreamClient implements AutoCloseable { private final WorkflowClient client; private final String workflowId; private final StreamPublisher publisher; + private final DataConverter itemDataConverter; @Nullable private final ScheduledExecutorService userPollExecutor; private final Map topicHandles = new HashMap<>(); @@ -83,14 +84,7 @@ private WorkflowStreamClient( this.client = client; this.workflowId = workflowId; - // A converter built only from PayloadConverters is codec-free, so items are never - // double-encoded against the codec on the client's signal/update envelope. - DataConverter dataConverter; - if (options.getPayloadConverters().length > 0) { - dataConverter = new DefaultDataConverter(options.getPayloadConverters()); - } else { - dataConverter = DefaultDataConverter.STANDARD_INSTANCE; - } + this.itemDataConverter = WorkflowStreamDataConverter.create(options.getPayloadConverters()); this.userPollExecutor = options.getPollExecutor(); @@ -98,7 +92,7 @@ private WorkflowStreamClient( this.publisher = new StreamPublisher( input -> stub.signal(WorkflowStreamConstants.PUBLISH_SIGNAL_NAME, input), - dataConverter, + itemDataConverter, options.getBatchInterval(), options.getMaxBatchSize(), options.getMaxRetryDuration()); @@ -112,6 +106,16 @@ public synchronized TopicHandle topic(String name) { return topicHandles.computeIfAbsent(name, n -> new TopicHandle(n, this)); } + /** Decodes an item using this client's configured, codec-free item converter. */ + public T decodeItem(WorkflowStreamItem item, Class valueClass) { + return decodeItem(item, valueClass, valueClass); + } + + /** Decodes an item using this client's configured, codec-free item converter. */ + public T decodeItem(WorkflowStreamItem item, Class valueClass, Type valueType) { + return itemDataConverter.fromPayload(item.getPayload(), valueClass, valueType); + } + /** * Sends buffered (and pending) items and waits for server confirmation. Returns once the items * buffered at call time have been signaled to the workflow and acknowledged. @@ -141,9 +145,9 @@ public long getOffset() { * } * *

The consuming thread blocks waiting for items; polling itself runs on the client's poll - * executor. Each item carries the raw {@link io.temporal.api.common.v1.Payload}; decode it with - * your data converter. The subscription ends cleanly when the workflow reaches a terminal state, - * automatically follows continue-as-new chains, and also ends when this client is closed. + * executor. Decode each item with {@link #decodeItem(WorkflowStreamItem, Class)}. The + * subscription ends cleanly when the workflow reaches a terminal state, automatically follows + * continue-as-new chains, and also ends when this client is closed. */ public WorkflowStreamSubscription subscribe(SubscribeOptions options) { return new WorkflowStreamSubscription(listener -> newSubscriptionDriver(options, listener)); diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java index ffd0d8e000..c4ba4e8504 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamClientOptions.java @@ -101,8 +101,9 @@ public Builder setMaxRetryDuration(Duration maxRetryDuration) { *

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

Default: the standard converter set. */ diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamDataConverter.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamDataConverter.java new file mode 100644 index 0000000000..d7e02b0eb3 --- /dev/null +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamDataConverter.java @@ -0,0 +1,27 @@ +package io.temporal.workflowstreams; + +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.PayloadConverter; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; + +/** + * Builds the converter used for individual Workflow Stream items. + * + *

Items are serialized into the stream protocol envelope, which is then sent through a Temporal + * signal or update. That outer envelope already uses the workflow client's or worker's configured + * data converter and payload codecs. Applying codecs to an item here would encode it a second time + * and prevent subscribers from decoding the raw item payload. This converter therefore includes + * only the configured payload converters, while still applying SDK-managed transfer conversion. + */ +final class WorkflowStreamDataConverter { + private WorkflowStreamDataConverter() {} + + static DataConverter create(PayloadConverter[] payloadConverters) { + DataConverter converter = + payloadConverters.length == 0 + ? DefaultDataConverter.STANDARD_INSTANCE + : new DefaultDataConverter(payloadConverters); + return TemporalTransferTypeDataConverter.wrap(converter); + } +} diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java index cf611fe641..9b1c7bddf9 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamItem.java @@ -4,10 +4,9 @@ import io.temporal.common.Experimental; /** - * A single decoded item yielded by a subscription. {@code payload} is the raw {@link Payload}; - * decode it at the call site with a payload converter, e.g. {@code - * DefaultDataConverter.STANDARD_INSTANCE.fromPayload(item.getPayload(), String.class, - * String.class)}. + * A single decoded item yielded by a subscription. {@code payload} is the raw {@link Payload}; use + * {@link WorkflowStreamClient#decodeItem(WorkflowStreamItem, Class)} to decode it with the stream + * client's configured item converter. */ @Experimental public final class WorkflowStreamItem { diff --git a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamOptions.java b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamOptions.java index 6c151537ec..94d5d71899 100644 --- a/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamOptions.java +++ b/contrib/temporal-workflowstreams/src/main/java/io/temporal/workflowstreams/WorkflowStreamOptions.java @@ -40,7 +40,8 @@ private Builder() {} *

As on the client side, only payload conversion happens here — never a payload codec. The * worker's codec chain runs once on the poll-update response that carries each batch to * subscribers, so encoding items here too would double-encode them; the {@code - * PayloadConverter[]} type makes that impossible. + * PayloadConverter[]} type makes that impossible. Transfer conversion and payload conversion + * apply to each item. * *

There is no public accessor for the worker's configured data converter inside workflow * code, so it cannot be picked up automatically; pass the matching payload converters here to diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java index 88c9270a7a..66403721a0 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTest.java @@ -5,6 +5,7 @@ import io.temporal.client.WorkflowStub; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.EncodingKeys; import io.temporal.testing.internal.SDKTestWorkflowRule; import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflow; import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflowImpl; @@ -69,6 +70,48 @@ public void testSubscribeDeliversItemsAndAdvancesOffset() { stub.getResult(Void.class); } + @Test + public void testClientPublishedItemUsesTransferConverter() { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + streamClient.topic("evt").publish(new TransferStreamTestModel("client"), true); + streamClient.flush(); + + try (WorkflowStreamSubscription subscription = streamClient.subscribe(FAST_POLL)) { + WorkflowStreamItem item = subscription.next(); + Assert.assertEquals( + "The configured payload converter must receive the protobuf transfer representation", + "json/protobuf", + item.getPayload() + .getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY) + .toStringUtf8()); + TransferStreamTestModel result = + streamClient.decodeItem(item, TransferStreamTestModel.class); + Assert.assertEquals(new TransferStreamTestModel("client"), result); + Assert.assertTrue(result.wasTransferred()); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + + @Test + public void testWorkflowPublishedItemUsesTransferConverter() { + WorkflowStub stub = startHostWorkflow(); + try (WorkflowStreamClient streamClient = newStreamClient(stub)) { + stub.signal("publishTransfer", "evt", new TransferStreamTestModel("workflow")); + + try (WorkflowStreamSubscription subscription = streamClient.subscribe(FAST_POLL)) { + TransferStreamTestModel result = + streamClient.decodeItem(subscription.next(), TransferStreamTestModel.class); + Assert.assertEquals(new TransferStreamTestModel("workflow"), result); + Assert.assertTrue(result.wasTransferred()); + } + } + stub.signal("finish"); + stub.getResult(Void.class); + } + @Test public void testTopicHandleSubscribeFilters() { WorkflowStub stub = startHostWorkflow(); diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java index 74b48bc938..d00cd93fbb 100644 --- a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/SubscribeTestWorkflows.java @@ -25,6 +25,9 @@ public interface SubscribeHostWorkflow { @SignalMethod void publishLocal(String topic, String value); + @SignalMethod + void publishTransfer(String topic, TransferStreamTestModel value); + @UpdateMethod void truncate(long upToOffset); } @@ -66,6 +69,11 @@ public void publishLocal(String topic, String value) { stream.topic(topic).publish(value); } + @Override + public void publishTransfer(String topic, TransferStreamTestModel value) { + stream.topic(topic).publish(value); + } + @Override public void truncate(long upToOffset) { stream.truncate(upToOffset); diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/TransferStreamTestModel.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/TransferStreamTestModel.java new file mode 100644 index 0000000000..91fe10644e --- /dev/null +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/TransferStreamTestModel.java @@ -0,0 +1,56 @@ +package io.temporal.workflowstreams; + +import com.google.protobuf.StringValue; +import io.temporal.common.converter.TransferTypeConverter; +import io.temporal.common.converter.TransferTypeConvertible; +import java.lang.reflect.Type; + +/** Test model that proves a Workflow Stream item uses transfer conversion. */ +@TransferTypeConvertible(TransferStreamTestModel.Converter.class) +public final class TransferStreamTestModel { + private final String value; + private final boolean transferred; + + public TransferStreamTestModel(String value) { + this(value, false); + } + + private TransferStreamTestModel(String value, boolean transferred) { + this.value = value; + this.transferred = transferred; + } + + public boolean wasTransferred() { + return transferred; + } + + @Override + public boolean equals(Object other) { + return other instanceof TransferStreamTestModel + && value.equals(((TransferStreamTestModel) other).value); + } + + @Override + public int hashCode() { + return value.hashCode(); + } + + public static final class Converter implements TransferTypeConverter { + public Converter() {} + + @Override + public Type getTransferType(Type valueType) { + return StringValue.class; + } + + @Override + public Object toTransferType(TransferStreamTestModel value) { + return StringValue.of(value.value); + } + + @Override + public TransferStreamTestModel fromTransferType(Object value, Type valueType) { + return new TransferStreamTestModel(((StringValue) value).getValue(), true); + } + } +} diff --git a/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java new file mode 100644 index 0000000000..99a4952a60 --- /dev/null +++ b/contrib/temporal-workflowstreams/src/test/java/io/temporal/workflowstreams/WorkflowStreamCodecTest.java @@ -0,0 +1,98 @@ +package io.temporal.workflowstreams; + +import com.google.protobuf.ByteString; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.common.converter.CodecDataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.payload.codec.PayloadCodec; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflow; +import io.temporal.workflowstreams.SubscribeTestWorkflows.SubscribeHostWorkflowImpl; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.Nonnull; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; + +public class WorkflowStreamCodecTest { + private static final String CODEC_METADATA_KEY = "workflow-stream-codec-test"; + private static final TrackingCodec CODEC = new TrackingCodec(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setDataConverter( + new CodecDataConverter( + DefaultDataConverter.newDefaultInstance(), + java.util.Collections.singletonList(CODEC))) + .build()) + .setWorkflowTypes(SubscribeHostWorkflowImpl.class) + .build(); + + @Test + public void codecsApplyToTheEnvelopeButNotIndividualItems() { + SubscribeHostWorkflow workflow = + testWorkflowRule.newWorkflowStubTimeoutOptions(SubscribeHostWorkflow.class); + WorkflowExecution execution = WorkflowClient.start(workflow::execute, null); + WorkflowStub stub = + testWorkflowRule.getWorkflowClient().newUntypedWorkflowStub(execution.getWorkflowId()); + CODEC.encodeCalls.set(0); + + try (WorkflowStreamClient client = + WorkflowStreamClient.newInstance( + testWorkflowRule.getWorkflowClient(), + execution.getWorkflowId(), + WorkflowStreamClientOptions.newBuilder() + .setBatchInterval(Duration.ofMillis(10)) + .build())) { + client.topic("events").publish("value", true); + client.flush(); + try (WorkflowStreamSubscription subscription = + client.subscribe(SubscribeOptions.getDefaultInstance())) { + WorkflowStreamItem item = subscription.next(); + Assert.assertFalse(item.getPayload().containsMetadata(CODEC_METADATA_KEY)); + Assert.assertEquals("value", client.decodeItem(item, String.class)); + } + } + Assert.assertTrue(CODEC.encodeCalls.get() > 0); + stub.signal("finish"); + stub.getResult(Void.class); + } + + private static final class TrackingCodec implements PayloadCodec { + private final AtomicInteger encodeCalls = new AtomicInteger(); + + @Override + @Nonnull + public List encode(@Nonnull List payloads) { + encodeCalls.incrementAndGet(); + List result = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + result.add( + payload.toBuilder() + .putMetadata(CODEC_METADATA_KEY, ByteString.copyFromUtf8("encoded")) + .build()); + } + return result; + } + + @Override + @Nonnull + public List decode(@Nonnull List payloads) { + List result = new ArrayList<>(payloads.size()); + for (Payload payload : payloads) { + result.add(payload.toBuilder().removeMetadata(CODEC_METADATA_KEY).build()); + } + return result; + } + } +} 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..22dc17d769 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityClientImpl.java @@ -11,6 +11,7 @@ import io.temporal.internal.client.RootActivityClientInvoker; import io.temporal.internal.client.external.GenericWorkflowClientImpl; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; import io.temporal.internal.util.MethodExtractor; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -37,6 +38,10 @@ class ActivityClientImpl implements ActivityClient, ActivityClientInternal { private final Scope metricsScope; ActivityClientImpl(WorkflowServiceStubs stubs, ActivityClientOptions options) { + options = + ActivityClientOptions.newBuilder(options) + .setDataConverter(TemporalTransferTypeDataConverter.wrap(options.getDataConverter())) + .build(); this.stubs = stubs; this.options = options; this.metricsScope = 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..2f89c13768 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/NexusClientImpl.java @@ -17,6 +17,7 @@ import io.temporal.internal.client.RootNexusClientInvoker; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.client.external.GenericWorkflowClientImpl; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import java.util.List; @@ -44,6 +45,12 @@ public static NexusClient newInstance(WorkflowServiceStubs service, NexusClientO } NexusClientImpl(WorkflowServiceStubs workflowServiceStubs, NexusClientResolvedOptions options) { + options = + new NexusClientResolvedOptions( + options.getNamespace(), + options.getInterceptors(), + TemporalTransferTypeDataConverter.wrap(options.getDataConverter()), + options.getIdentity()); workflowServiceStubs = new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace()); this.workflowServiceStubs = workflowServiceStubs; 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..1ed094f174 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java @@ -14,6 +14,7 @@ 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; @@ -23,6 +24,7 @@ import io.temporal.internal.client.external.GenericWorkflowClientImpl; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; import io.temporal.internal.common.PluginUtils; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.sync.StubMarker; import io.temporal.internal.worker.HeartbeatManager; @@ -52,6 +54,8 @@ final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClient private final GenericWorkflowClient genericClient; private final WorkflowClientOptions options; + private final WorkflowClientOptions internalOptions; + private final DataConverter internalDataConverter; private final ManualActivityCompletionClientFactory manualActivityCompletionClientFactory; private final WorkflowClientCallsInterceptor workflowClientCallsInvoker; private final WorkflowServiceStubs workflowServiceStubs; @@ -103,9 +107,12 @@ public static WorkflowClient newInstance( // Set merged plugins after configuration, then validate builder.setPlugins(mergedPlugins); options = builder.validateAndBuildWithDefaults(); + this.internalDataConverter = TemporalTransferTypeDataConverter.wrap(options.getDataConverter()); workflowServiceStubs = new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace()); this.options = options; + this.internalOptions = + WorkflowClientOptions.newBuilder(options).setDataConverter(internalDataConverter).build(); this.workflowServiceStubs = workflowServiceStubs; this.metricsScope = workflowServiceStubs @@ -124,7 +131,7 @@ public static WorkflowClient newInstance( workflowServiceStubs, options.getNamespace(), options.getIdentity(), - options.getDataConverter(), + getInternalDataConverter(), externalStorageRunner); java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval(); @@ -142,7 +149,7 @@ public static WorkflowClient newInstance( private WorkflowClientCallsInterceptor initializeClientInvoker() { WorkflowClientCallsInterceptor workflowClientInvoker = new RootWorkflowClientInvoker( - genericClient, options, workerFactoryRegistry, externalStorageRunner); + genericClient, internalOptions, workerFactoryRegistry, externalStorageRunner); for (WorkflowClientInterceptor clientInterceptor : interceptors) { workflowClientInvoker = clientInterceptor.workflowClientCallsInterceptor(workflowClientInvoker); @@ -160,13 +167,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, internalOptions, workflowClientCallsInvoker, options); return (T) Proxy.newProxyInstance( workflowInterface.getClassLoader(), @@ -243,7 +255,7 @@ public T newWorkflowStub( WorkflowInvocationHandler invocationHandler = new WorkflowInvocationHandler( workflowInterface, - this.getOptions(), + internalOptions, workflowClientCallsInvoker, execution.build(), legacyTargeting, @@ -267,7 +279,8 @@ public WorkflowStub newUntypedWorkflowStub(String workflowId) { @SuppressWarnings("deprecation") public WorkflowStub newUntypedWorkflowStub(String workflowType, WorkflowOptions workflowOptions) { WorkflowStub result = - new WorkflowStubImpl(options, workflowClientCallsInvoker, workflowType, workflowOptions); + new WorkflowStubImpl( + internalOptions, workflowClientCallsInvoker, workflowType, workflowOptions); for (WorkflowClientInterceptor i : interceptors) { result = i.newUntypedWorkflowStub(workflowType, workflowOptions, result); } @@ -319,7 +332,7 @@ WorkflowStub newUntypedWorkflowStub( } WorkflowStub result = new WorkflowStubImpl( - options, + internalOptions, workflowClientCallsInvoker, workflowType, execution.build(), 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..3bb931ee73 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 @@ -11,6 +11,7 @@ import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.client.external.GenericWorkflowClientImpl; import io.temporal.internal.common.PluginUtils; +import io.temporal.internal.common.converter.TemporalTransferTypeDataConverter; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.WorkflowServiceStubsPlugin; @@ -71,6 +72,10 @@ public static ScheduleClient newInstance( // Set merged plugins after configuration, then build builder.setPlugins(mergedPlugins); options = builder.build(); + options = + ScheduleClientOptions.newBuilder(options) + .setDataConverter(TemporalTransferTypeDataConverter.wrap(options.getDataConverter())) + .build(); workflowServiceStubs = new NamespaceInjectWorkflowServiceStubs(workflowServiceStubs, options.getNamespace()); 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/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..5899f7b35c --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverter.java @@ -0,0 +1,287 @@ +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); + // Prevent a converter from returning a value that cannot be passed to the requested model type. + return valueClass.cast(fromTransferValue(value, descriptor, requestedType)); + } + + @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); + // Prevent a converter from returning a value that cannot be passed to the requested model type. + return valueClass.cast(fromTransferValue(value, descriptor, requestedType)); + } + + @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); + } + } + 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 Object fromTransferValue( + Object value, ConverterDescriptor descriptor, Type valueType) { + return descriptor.converter().fromTransferType(value, valueType); + } + + 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 6627e3fd99..d329717f60 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -129,10 +129,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/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 395a0be2bf..d3ae478c6a 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.internal.SDKTestWorkflowRule; import io.temporal.workflow.shared.EchoNexusServiceImpl; import io.temporal.workflow.shared.TestNexusServices; @@ -36,6 +39,7 @@ public class NexusServiceClientTest { .setWorkflowTypes(PlaceholderWorkflowImpl.class) .setNexusServiceImplementation( new EchoNexusServiceImpl(), + new TransferServiceImpl(), new VoidInputServiceImpl(), new VoidReturnServiceImpl(), new VoidServiceImpl()) @@ -59,6 +63,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 = @@ -218,6 +234,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 f7f5f50daf..6bb79d7193 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.internal.SDKTestWorkflowRule; @@ -116,6 +118,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 d91922884a..463efaa8b1 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/common/PluginPropagationTest.java @@ -22,13 +22,21 @@ import static org.junit.Assert.*; +import com.google.protobuf.StringValue; +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.TransferTypeConverter; +import io.temporal.common.converter.TransferTypeConvertible; +import io.temporal.internal.client.WorkflowClientInternal; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.testing.TestEnvironmentOptions; import io.temporal.testing.TestWorkflowEnvironment; 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; @@ -41,6 +49,68 @@ */ 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()); + assertSame(plugin.dataConverter, env.getWorkflowClient().getOptions().getDataConverter()); + Payload payload = + ((WorkflowClientInternal) env.getWorkflowClient().getInternal()) + .getInternalDataConverter() + .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/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..60513a064f --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/functional/serialization/TransferTypeIntegrationTest.java @@ -0,0 +1,126 @@ +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 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/common/converter/TemporalTransferTypeDataConverterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverterTest.java new file mode 100644 index 0000000000..46962169b6 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/common/converter/TemporalTransferTypeDataConverterTest.java @@ -0,0 +1,678 @@ +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.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterException; +import io.temporal.common.converter.DefaultDataConverter; +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.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 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 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())); + } + + @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(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/test/java/io/temporal/testing/TestActivityEnvironmentTransferTypeTest.java b/temporal-testing/src/test/java/io/temporal/testing/TestActivityEnvironmentTransferTypeTest.java new file mode 100644 index 0000000000..2ec7d62705 --- /dev/null +++ b/temporal-testing/src/test/java/io/temporal/testing/TestActivityEnvironmentTransferTypeTest.java @@ -0,0 +1,279 @@ +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 activityContextExposesConfiguredConverter() { + TrackingDataConverter trackingConverter = new TrackingDataConverter(); + TestActivityEnvironment environment = newEnvironment(trackingConverter); + try { + environment.registerActivitiesImplementations(new ConverterActivityImpl(trackingConverter)); + ConverterActivity activity = environment.newActivityStub(ConverterActivity.class); + + assertTrue(activity.usesConfiguredConverter()); + } 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 usesConfiguredConverter(); + } + + private static final class ConverterActivityImpl implements ConverterActivity { + private final DataConverter expectedConverter; + + private ConverterActivityImpl(DataConverter expectedConverter) { + this.expectedConverter = expectedConverter; + } + + @Override + public boolean usesConfiguredConverter() { + return Activity.getExecutionContext().getWorkflowClient().getOptions().getDataConverter() + == expectedConverter; + } + } + + @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); + } + } +}