diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index fc85f0039f..55297194fe 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -32,6 +32,7 @@ import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; +import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.serviceclient.StatusUtils; import java.lang.reflect.Type; import java.util.*; @@ -61,7 +62,18 @@ public StartActivityOutput startActivity(StartActivityInput input) { if (Strings.isNullOrEmpty(options.getTaskQueue())) { throw new IllegalArgumentException("taskQueue must not be null or empty"); } - DataConverter dc = clientOptions.getDataConverter(); + DataConverter dc = + clientOptions + .getDataConverter() + .withContext( + new ActivitySerializationContext( + clientOptions.getNamespace(), + null, + null, + input.getActivityType(), + options.getTaskQueue(), + false)); + InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null; NexusOperationMetadata nexusOperationMetadata = @@ -201,7 +213,8 @@ public StartActivityOutput startActivity(StartActivityInput input) { public GetActivityResultOutput getActivityResult(GetActivityResultInput input) throws TimeoutException { String namespace = clientOptions.getNamespace(); - DataConverter dc = clientOptions.getDataConverter(); + DataConverter dc = + clientOptions.getDataConverter().withContext(resultSerializationContext(input)); Deadline deadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit()); while (true) { @@ -276,7 +289,8 @@ public GetActivityResultOutput getActivityResult(GetActivityResultInput CompletableFuture> getActivityResultAsync( GetActivityResultInput input) { - DataConverter dc = clientOptions.getDataConverter(); + DataConverter dc = + clientOptions.getDataConverter().withContext(resultSerializationContext(input)); Deadline deadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit()); return pollActivityUntilOutcome(input, deadline) .handle( @@ -358,6 +372,13 @@ private GetActivityResultOutput decodeOutcome( } } + private ActivitySerializationContext resultSerializationContext(GetActivityResultInput input) { + // Currently, result doesn't have access to activity type and serialization context doesn't hold + // activity ID. + return new ActivitySerializationContext( + clientOptions.getNamespace(), null, null, null, null, false); + } + @Override public DescribeActivityOutput describeActivity(DescribeActivityInput input) { DescribeActivityExecutionRequest.Builder req = diff --git a/temporal-sdk/src/main/java/io/temporal/payload/context/ActivitySerializationContext.java b/temporal-sdk/src/main/java/io/temporal/payload/context/ActivitySerializationContext.java index 57dcfb3138..78ba63d247 100644 --- a/temporal-sdk/src/main/java/io/temporal/payload/context/ActivitySerializationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/payload/context/ActivitySerializationContext.java @@ -11,32 +11,34 @@ public class ActivitySerializationContext implements HasWorkflowSerializationCon private final @Nonnull String namespace; private final @Nullable String workflowId; private final @Nullable String workflowType; - private final @Nonnull String activityType; - private final @Nonnull String activityTaskQueue; + private final @Nullable String activityType; + private final @Nullable String activityTaskQueue; private final boolean local; /** * @param namespace the activity's namespace; must not be {@code null} * @param workflowId the workflow ID that scheduled the activity, or {@code null} for standalone - * activities (stored as an empty string) + * activities * @param workflowType the workflow type that scheduled the activity, or {@code null} for - * standalone activities (stored as an empty string) - * @param activityType the activity type name; must not be {@code null} - * @param activityTaskQueue the task queue for this activity; must not be {@code null} + * standalone activities + * @param activityType the activity type name, or {@code null} if unknown. Activity type is + * unknown when getting a Standalone Activity result. + * @param activityTaskQueue the task queue for this activity, or {@code null} if unknown. Task + * queue is unknown when getting a Standalone Activity result. * @param local {@code true} if this is a local activity */ public ActivitySerializationContext( @Nonnull String namespace, @Nullable String workflowId, @Nullable String workflowType, - @Nonnull String activityType, - @Nonnull String activityTaskQueue, + @Nullable String activityType, + @Nullable String activityTaskQueue, boolean local) { this.namespace = Objects.requireNonNull(namespace); this.workflowId = workflowId; this.workflowType = workflowType; - this.activityType = Objects.requireNonNull(activityType); - this.activityTaskQueue = Objects.requireNonNull(activityTaskQueue); + this.activityType = activityType; + this.activityTaskQueue = activityTaskQueue; this.local = local; } @@ -67,12 +69,12 @@ public String getWorkflowType() { return workflowType; } - @Nonnull + @Nullable public String getActivityType() { return activityType; } - @Nonnull + @Nullable public String getActivityTaskQueue() { return activityTaskQueue; } diff --git a/temporal-sdk/src/test/java/io/temporal/payload/context/ContextAwareDataConverterTest.java b/temporal-sdk/src/test/java/io/temporal/payload/context/ContextAwareDataConverterTest.java new file mode 100644 index 0000000000..8e6c45c3fb --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/payload/context/ContextAwareDataConverterTest.java @@ -0,0 +1,534 @@ +package io.temporal.payload.context; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.ActivityOptions; +import io.temporal.activity.LocalActivityOptions; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.client.*; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DataConverterException; +import io.temporal.common.converter.GlobalDataConverter; +import io.temporal.internal.history.LocalActivityMarkerUtils; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.lang.reflect.Type; +import java.time.Duration; +import java.util.*; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.jspecify.annotations.NonNull; +import org.jspecify.annotations.Nullable; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; + +public class ContextAwareDataConverterTest { + @ActivityInterface + public interface Activities { + @ActivityMethod(name = "HelloActivity") + TracedValue hello(TracedValue input); + } + + public static class ActivitiesImpl implements Activities { + @Override + public TracedValue hello(TracedValue input) { + return new TracedValue("Hello " + input.getValue()); + } + } + + @WorkflowInterface + public interface HelloWorkflow { + @WorkflowMethod + TracedValue execute(TracedValue arg, boolean local); + } + + public static class HelloWorkflowImpl implements HelloWorkflow { + private final Activities activities = + Workflow.newActivityStub( + Activities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + private final Activities localActivities = + Workflow.newLocalActivityStub( + Activities.class, + LocalActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build()); + + @Override + public TracedValue execute(TracedValue arg, boolean local) { + if (local) { + return localActivities.hello(arg); + } else { + return activities.hello(arg); + } + } + } + + private static final String TAG_WORKER = "worker"; + private static final String TAG_CLIENT = "client"; + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(HelloWorkflowImpl.class) + .setActivityImplementations(new ActivitiesImpl()) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setDataConverter(new TracingDataConverter(GlobalDataConverter.get(), TAG_WORKER)) + .build()) + .setActivityClientOptions( + ActivityClientOptions.newBuilder() + .setNamespace(SDKTestWorkflowRule.NAMESPACE) + .setDataConverter(new TracingDataConverter(GlobalDataConverter.get(), TAG_CLIENT)) + .build()) + .build(); + + @Test + public void standaloneActivitySerializationContext() { + String activityId = "act-" + UUID.randomUUID(); + + ActivityHandle handle = + testWorkflowRule + .getActivityClient() + .start( + Activities.class, + Activities::hello, + StartActivityOptions.newBuilder() + .setId(activityId) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(), + new TracedValue("world")); + + ActivitySerializationContext workerContext = + new ActivitySerializationContext( + SDKTestWorkflowRule.NAMESPACE, + null, + null, + "HelloActivity", + testWorkflowRule.getTaskQueue(), + false); + + // Currently, client doesn't set activityType in serialization context + ActivitySerializationContext clientContext = + new ActivitySerializationContext( + SDKTestWorkflowRule.NAMESPACE, + null, + null, + null, + testWorkflowRule.getTaskQueue(), + false); + + TracedValue expected = + new TracedValue("Hello world") + .addTrace( + TraceEntry.encode(workerContext, TAG_WORKER), + TraceEntry.decode(clientContext, TAG_CLIENT)); + + Assert.assertEquals(expected, handle.getResult()); + Assert.assertEquals(expected, handle.getResultAsync().join()); + } + + @Test + public void workflowActivitySerializationContext() { + WorkflowClient client = getWorkflowClient(); + + HelloWorkflow workflow = + client.newWorkflowStub( + HelloWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowRunTimeout(Duration.ofSeconds(10)) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + + TracedValue result = workflow.execute(new TracedValue("world"), false); + WorkflowExecution execution = WorkflowStub.fromTyped(workflow).getExecution(); + Assert.assertNotNull(execution); + List history = + client + .fetchHistory(execution.getWorkflowId(), execution.getRunId()) + .getHistory() + .getEventsList(); + List scheduledEvents = + history.stream() + .filter(HistoryEvent::hasActivityTaskScheduledEventAttributes) + .collect(Collectors.toList()); + Assert.assertEquals(1, scheduledEvents.size()); + HistoryEvent scheduledEvent = scheduledEvents.get(0); + List completedEvents = + history.stream() + .filter(HistoryEvent::hasActivityTaskCompletedEventAttributes) + .collect(Collectors.toList()); + Assert.assertEquals(1, completedEvents.size()); + HistoryEvent completedEvent = completedEvents.get(0); + Assert.assertEquals( + scheduledEvent.getEventId(), + completedEvent.getActivityTaskCompletedEventAttributes().getScheduledEventId()); + + WorkflowSerializationContext workflowContext = + new WorkflowSerializationContext(SDKTestWorkflowRule.NAMESPACE, execution.getWorkflowId()); + + ActivitySerializationContext activityContext = + new ActivitySerializationContext( + SDKTestWorkflowRule.NAMESPACE, + execution.getWorkflowId(), + "HelloWorkflow", + "HelloActivity", + testWorkflowRule.getTaskQueue(), + false); + + Assert.assertEquals( + new TracedValue("Hello world") + .addTrace( + TraceEntry.encode(activityContext, TAG_WORKER), + TraceEntry.decode(activityContext, TAG_WORKER), + TraceEntry.encode(workflowContext, TAG_WORKER), + TraceEntry.decode(workflowContext, TAG_CLIENT)), + result); + + Assert.assertEquals( + new TracedValue("world") + .addTrace( + TraceEntry.encode(workflowContext, TAG_CLIENT), + TraceEntry.decode(workflowContext, TAG_WORKER), + TraceEntry.encode(activityContext, TAG_WORKER)), + GlobalDataConverter.get() + .fromPayloads( + 0, + Optional.of(scheduledEvent.getActivityTaskScheduledEventAttributes().getInput()), + TracedValue.class, + TracedValue.class)); + + Assert.assertEquals( + new TracedValue("Hello world").addTrace(TraceEntry.encode(activityContext, TAG_WORKER)), + GlobalDataConverter.get() + .fromPayloads( + 0, + Optional.of(completedEvent.getActivityTaskCompletedEventAttributes().getResult()), + TracedValue.class, + TracedValue.class)); + } + + @Test + public void localActivitySerializationContext() { + WorkflowClient client = getWorkflowClient(); + + HelloWorkflow workflow = + client.newWorkflowStub( + HelloWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowRunTimeout(Duration.ofSeconds(10)) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .build()); + + TracedValue result = workflow.execute(new TracedValue("world"), true); + WorkflowExecution execution = WorkflowStub.fromTyped(workflow).getExecution(); + Assert.assertNotNull(execution); + List history = + client + .fetchHistory(execution.getWorkflowId(), execution.getRunId()) + .getHistory() + .getEventsList(); + List markerEvents = + history.stream() + .filter(HistoryEvent::hasMarkerRecordedEventAttributes) + .collect(Collectors.toList()); + Assert.assertEquals(1, markerEvents.size()); + HistoryEvent markerEvent = markerEvents.get(0); + Assert.assertTrue(LocalActivityMarkerUtils.hasLocalActivityStructure(markerEvent)); + + WorkflowSerializationContext workflowContext = + new WorkflowSerializationContext(SDKTestWorkflowRule.NAMESPACE, execution.getWorkflowId()); + + ActivitySerializationContext activityContext = + new ActivitySerializationContext( + SDKTestWorkflowRule.NAMESPACE, + execution.getWorkflowId(), + "HelloWorkflow", + "HelloActivity", + testWorkflowRule.getTaskQueue(), + true); + + Assert.assertEquals( + new TracedValue("Hello world") + .addTrace( + TraceEntry.encode(activityContext, TAG_WORKER), + TraceEntry.decode(activityContext, TAG_WORKER), + TraceEntry.encode(workflowContext, TAG_WORKER), + TraceEntry.decode(workflowContext, TAG_CLIENT)), + result); + + Assert.assertEquals( + new TracedValue("Hello world").addTrace(TraceEntry.encode(activityContext, TAG_WORKER)), + GlobalDataConverter.get() + .fromPayloads( + 0, + Optional.ofNullable( + LocalActivityMarkerUtils.getResult( + markerEvent.getMarkerRecordedEventAttributes())), + TracedValue.class, + TracedValue.class)); + } + + private WorkflowClient getWorkflowClient() { + WorkflowClient client = testWorkflowRule.getWorkflowClient(); + WorkflowClientOptions options = + client.getOptions().toBuilder() + .setDataConverter(new TracingDataConverter(GlobalDataConverter.get(), TAG_CLIENT)) + .build(); + return WorkflowClient.newInstance(client.getWorkflowServiceStubs(), options); + } + + private static class TracingDataConverter implements DataConverter { + private final DataConverter dc; + private final String tag; + private final SerializationContext context; + + public TracingDataConverter(DataConverter dc, String tag) { + this(dc, tag, null); + } + + private TracingDataConverter(DataConverter dc, String tag, SerializationContext context) { + this.dc = dc; + this.tag = tag; + this.context = context; + } + + @Override + public Optional toPayload(T value) throws DataConverterException { + if (value instanceof TracedValue) { + return dc.toPayload(((TracedValue) value).addTrace(TraceEntry.encode(context, tag))); + } else { + return dc.toPayload(value); + } + } + + @Override + public T fromPayload(Payload payload, Class valueClass, Type valueType) + throws DataConverterException { + if (valueClass == TracedValue.class) { + return valueClass.cast( + dc.fromPayload(payload, TracedValue.class, valueType) + .addTrace(TraceEntry.decode(context, tag))); + } else { + return dc.fromPayload(payload, valueClass, valueType); + } + } + + @Override + public Optional toPayloads(Object... values) throws DataConverterException { + if (Arrays.stream(values).anyMatch(v -> v instanceof TracedValue)) { + Payloads.Builder builder = Payloads.newBuilder(); + for (Object v : values) { + builder.addPayloads(toPayload(v).get()); + } + return Optional.of(builder.build()); + } else { + return dc.toPayloads(values); + } + } + + @Override + public T fromPayloads( + int index, Optional content, Class valueType, Type valueGenericType) + throws DataConverterException { + if (valueType == TracedValue.class) { + return valueType.cast( + dc.fromPayloads(index, content, TracedValue.class, valueGenericType) + .addTrace(TraceEntry.decode(context, tag))); + } else { + return dc.fromPayloads(index, content, valueType, valueGenericType); + } + } + + @Override + public @NonNull DataConverter withContext(@NonNull SerializationContext context) { + return new TracingDataConverter(dc, tag, context); + } + } + + public static class TracedValue { + private final String value; + private final ArrayList trace; + + public TracedValue(String value) { + this.value = value; + this.trace = new ArrayList<>(); + } + + @JsonCreator + public TracedValue( + @JsonProperty("value") String value, @JsonProperty("trace") List trace) { + this.value = value; + this.trace = new ArrayList<>(trace); + } + + public String getValue() { + return value; + } + + public List getTrace() { + return Collections.unmodifiableList(trace); + } + + public TracedValue addTrace(TraceEntry... entries) { + return new TracedValue( + value, + Stream.concat(trace.stream(), Arrays.stream(entries)).collect(Collectors.toList())); + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) return false; + TracedValue that = (TracedValue) o; + return Objects.equals(value, that.value) && Objects.equals(trace, that.trace); + } + + @Override + public int hashCode() { + return Objects.hash(value, trace); + } + + @Override + public String toString() { + return "TracedValue{" + "value='" + value + '\'' + ", trace=" + trace + '}'; + } + } + + public static class TraceEntry { + private final @NonNull String tag; + private final Operation operation; + private final @Nullable String namespace; + private final @Nullable String workflowId; + private final @Nullable String activityType; + private final @Nullable Boolean local; + + @JsonCreator + public TraceEntry( + @JsonProperty("tag") @NonNull String tag, + @JsonProperty("operation") Operation operation, + @JsonProperty("namespace") @Nullable String namespace, + @JsonProperty("workflowId") @Nullable String workflowId, + @JsonProperty("activityType") @Nullable String activityType, + @JsonProperty("local") @Nullable Boolean local) { + this.tag = tag; + this.operation = operation; + this.namespace = namespace; + this.workflowId = workflowId; + this.activityType = activityType; + this.local = local; + } + + public TraceEntry( + @Nullable SerializationContext context, @NonNull String tag, Operation operation) { + this.tag = tag; + this.operation = operation; + if (context == null) { + namespace = null; + workflowId = null; + activityType = null; + local = null; + } else if (context instanceof WorkflowSerializationContext) { + WorkflowSerializationContext c = (WorkflowSerializationContext) context; + namespace = c.getNamespace(); + workflowId = c.getWorkflowId(); + activityType = null; + local = null; + } else if (context instanceof ActivitySerializationContext) { + ActivitySerializationContext c = (ActivitySerializationContext) context; + namespace = c.getNamespace(); + workflowId = c.getWorkflowId(); + activityType = c.getActivityType(); + local = c.isLocal(); + } else { + throw new IllegalArgumentException( + "Unknown context type: " + context.getClass().getCanonicalName()); + } + } + + public static TraceEntry encode(@NonNull SerializationContext context, @NonNull String tag) { + return new TraceEntry(context, tag, Operation.Encode); + } + + public static TraceEntry decode(@NonNull SerializationContext context, @NonNull String tag) { + return new TraceEntry(context, tag, Operation.Decode); + } + + public @NonNull String getTag() { + return tag; + } + + public Operation getOperation() { + return operation; + } + + public @Nullable String getNamespace() { + return namespace; + } + + public @Nullable String getWorkflowId() { + return workflowId; + } + + public @Nullable String getActivityType() { + return activityType; + } + + public @Nullable Boolean isLocal() { + return local; + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) return false; + TraceEntry that = (TraceEntry) o; + return Objects.equals(tag, that.tag) + && operation == that.operation + && Objects.equals(namespace, that.namespace) + && Objects.equals(workflowId, that.workflowId) + && Objects.equals(activityType, that.activityType) + && Objects.equals(local, that.local); + } + + @Override + public int hashCode() { + return Objects.hash(tag, operation, namespace, workflowId, activityType, local); + } + + @Override + public String toString() { + return "TraceEntry{" + + "tag='" + + tag + + '\'' + + ", operation=" + + operation + + ", namespace='" + + namespace + + '\'' + + ", workflowId='" + + workflowId + + '\'' + + ", activityType='" + + activityType + + '\'' + + ", local=" + + local + + '}'; + } + + public enum Operation { + Encode, + Decode + } + } +}