diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d5b62403e..8ba32db9f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,9 @@ to docs, or any other relevant information. ## [Unreleased] ### Added +- Added experimental `Copyable` for isolated in-process copies. Local activities copy + eligible inputs for each attempt and outputs for live execution, preserving history serialization + and replay. Deeply immutable values may return themselves. - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure` diff --git a/temporal-sdk/src/main/java/io/temporal/common/Copyable.java b/temporal-sdk/src/main/java/io/temporal/common/Copyable.java new file mode 100644 index 0000000000..7113bb1db5 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/common/Copyable.java @@ -0,0 +1,30 @@ +package io.temporal.common; + +/** + * Supports isolated in-process value transfer without a serialization round trip. + * + *

Copies must not share mutable reachable state with the original. Deeply immutable values may + * return {@code this}. A copy must be non-null and compatible with the original's runtime type. + * + *

Copying must be side-effect-free, safe for concurrent calls, and deterministically equivalent + * to a round trip through the configured data converter. The SDK may call it on workflow or + * activity threads, including during replay. Object identity must not affect workflow behavior. + * + *

Currently, only local activities use this interface. Input batches use scheduling-time + * snapshots when every non-null argument is copyable. Compatible requested types use fresh copies; + * other types use the data converter. Copying is independent of marker options; omitted inputs + * avoid serialization unless conversion or explicit payload access requires it. Outputs remain + * serialized for history and deserialized on replay. Other execution paths, including eager + * workflows and eager activities, do not yet use this interface. + * + * @param the type of the copied value. + */ +@Experimental +public interface Copyable> { + /** + * Returns an independent copy satisfying this interface's contract. + * + * @return a non-null copy, or {@code this} for a deeply immutable value. + */ + T copy(); +} diff --git a/temporal-sdk/src/main/java/io/temporal/common/converter/EncodedValues.java b/temporal-sdk/src/main/java/io/temporal/common/converter/EncodedValues.java index 1662d8fd99..2a432ef217 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/converter/EncodedValues.java +++ b/temporal-sdk/src/main/java/io/temporal/common/converter/EncodedValues.java @@ -5,7 +5,7 @@ import java.util.Objects; import java.util.Optional; -public final class EncodedValues implements Values { +public class EncodedValues implements Values { private Optional payloads; private DataConverter converter; private final Object[] values; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java index 2d7b83c274..5ad394e41d 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoImpl.java @@ -15,6 +15,7 @@ import java.util.Objects; import java.util.Optional; import javax.annotation.Nonnull; +import javax.annotation.Nullable; final class ActivityInfoImpl implements ActivityInfoInternal { private final String namespace; @@ -22,18 +23,21 @@ final class ActivityInfoImpl implements ActivityInfoInternal { private final PollActivityTaskQueueResponseOrBuilder response; private final boolean local; private final Functions.Proc completionHandle; + private final ActivityInput input; ActivityInfoImpl( PollActivityTaskQueueResponseOrBuilder response, @Nonnull String namespace, @Nonnull String activityTaskQueue, boolean local, - Functions.Proc completionHandle) { + Functions.Proc completionHandle, + @Nullable LocalActivityArguments snapshot) { this.response = Objects.requireNonNull(response); this.namespace = Objects.requireNonNull(namespace); this.activityTaskQueue = Objects.requireNonNull(activityTaskQueue); this.local = local; this.completionHandle = completionHandle; + this.input = new ActivityInput(response, snapshot); } @Override @@ -190,11 +194,8 @@ public Functions.Proc getCompletionHandle() { } @Override - public Optional getInput() { - if (response.hasInput()) { - return Optional.of(response.getInput()); - } - return Optional.empty(); + public ActivityInput getInput() { + return input; } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoInternal.java index aad6b1ad20..d6e82ef38a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInfoInternal.java @@ -2,7 +2,6 @@ import io.temporal.activity.ActivityInfo; import io.temporal.api.common.v1.Header; -import io.temporal.api.common.v1.Payloads; import io.temporal.workflow.Functions; import java.util.Optional; @@ -21,7 +20,7 @@ interface ActivityInfoInternal extends ActivityInfo { /** * @return input parameters of the activity execution */ - Optional getInput(); + ActivityInput getInput(); /** * @return header that is passed with the activity execution diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInput.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInput.java new file mode 100644 index 0000000000..4d8ac9098d --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityInput.java @@ -0,0 +1,81 @@ +package io.temporal.internal.activity; + +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponseOrBuilder; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.EncodedValues; +import java.lang.reflect.Type; +import java.util.Objects; +import java.util.Optional; +import javax.annotation.Nullable; + +/** One source of activity arguments: serialized payloads or a captured copyable snapshot. */ +final class ActivityInput { + // A cached serialized view of the input, materialized only if needed. + private @Nullable Optional payloads; + private final @Nullable LocalActivityArguments snapshot; + + ActivityInput( + PollActivityTaskQueueResponseOrBuilder response, @Nullable LocalActivityArguments snapshot) { + this.snapshot = snapshot; + this.payloads = + response.hasInput() + ? Optional.of(response.getInput()) + : snapshot == null ? Optional.empty() : null; + } + + Object[] getArguments(Class[] types, Type[] genericTypes, DataConverter converter) { + if (snapshot != null && snapshot.canCopyTo(types, genericTypes)) { + return snapshot.copy(); + } + return converter.fromPayloads(toPayloads(converter), types, genericTypes); + } + + private Optional toPayloads(DataConverter converter) { + if (payloads == null) { + payloads = snapshot.toPayloads(converter); + } + return payloads; + } + + EncodedValues getEncodedValues(DataConverter converter) { + if (snapshot == null) { + return new EncodedValues(payloads, converter); + } + return new CopyableEncodedValues(converter); + } + + /** + * Reads independent copies without exposing the snapshot used by conversion and serialization. + */ + private final class CopyableEncodedValues extends EncodedValues { + private DataConverter converter; + + private CopyableEncodedValues(DataConverter converter) { + setDataConverter(converter); + } + + @Override + public int getSize() { + return snapshot.size(); + } + + @Override + public Optional toPayloads() { + return ActivityInput.this.toPayloads(converter); + } + + @Override + public void setDataConverter(DataConverter converter) { + this.converter = Objects.requireNonNull(converter); + } + + @Override + public T get(int index, Class type, Type genericType) { + if (snapshot.canCopyTo(index, type, genericType)) { + return type.cast(snapshot.copy(index)); + } + return converter.fromPayloads(index, toPayloads(), type, genericType); + } + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityPollResponseToInfo.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityPollResponseToInfo.java index d6d1d1e197..a449022538 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityPollResponseToInfo.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityPollResponseToInfo.java @@ -9,6 +9,6 @@ public static ActivityInfo toActivityInfoImpl( String namespace, String activityTaskQueue, boolean local) { - return new ActivityInfoImpl(response, namespace, activityTaskQueue, local, null); + return new ActivityInfoImpl(response, namespace, activityTaskQueue, local, null, null); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java index 7789bd2716..db0edd9d85 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java @@ -9,13 +9,14 @@ import io.temporal.api.common.v1.Payloads; import io.temporal.api.workflowservice.v1.RespondActivityTaskCompletedRequest; import io.temporal.client.ActivityCanceledException; +import io.temporal.common.Copyable; import io.temporal.common.context.ContextPropagator; import io.temporal.common.converter.DataConverter; -import io.temporal.common.converter.EncodedValues; import io.temporal.common.interceptors.ActivityInboundCallsInterceptor; import io.temporal.common.interceptors.ActivityInboundCallsInterceptor.ActivityOutput; import io.temporal.common.interceptors.Header; import io.temporal.common.interceptors.WorkerInterceptor; +import io.temporal.internal.common.CopyableUtils; import io.temporal.internal.common.FailureUtils; import io.temporal.internal.worker.ActivityTaskHandler; import io.temporal.payload.context.ActivitySerializationContext; @@ -147,7 +148,7 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri abstract Object getActivity(); abstract Object[] provideArgs( - Optional input, DataConverter dataConverterWithActivityContext); + ActivityInput input, DataConverter dataConverterWithActivityContext); protected abstract ActivityTaskHandler.Result constructSuccessfulResultValue( ActivityInfoInternal info, @@ -160,13 +161,19 @@ ActivityTaskHandler.Result constructResultValue( DataConverter dataConverterWithActivityContext) { RespondActivityTaskCompletedRequest.Builder request = RespondActivityTaskCompletedRequest.newBuilder(); + Object copiedResult = null; if (result != null) { - Optional serialized = - dataConverterWithActivityContext.toPayloads(result.getResult()); + Object value = result.getResult(); + if (info.isLocal() && result.getResult() instanceof Copyable) { + copiedResult = CopyableUtils.copy(value); + value = copiedResult; + } + // Serialize the same snapshot delivered to the workflow, not the activity's retained value. + Optional serialized = dataConverterWithActivityContext.toPayloads(value); serialized.ifPresent(request::setResult); } return new ActivityTaskHandler.Result( - info.getActivityId(), request.build(), null, null, false); + info.getActivityId(), request.build(), null, null, false, copiedResult); } static void deserializeAndPopulateContext( @@ -211,9 +218,11 @@ Object getActivity() { } @Override - Object[] provideArgs(Optional input, DataConverter dataConverterWithActivityContext) { - return dataConverterWithActivityContext.fromPayloads( - input, method.getParameterTypes(), method.getGenericParameterTypes()); + Object[] provideArgs(ActivityInput input, DataConverter dataConverterWithActivityContext) { + return input.getArguments( + method.getParameterTypes(), + method.getGenericParameterTypes(), + dataConverterWithActivityContext); } @Override @@ -254,9 +263,8 @@ Object getActivity() { } @Override - Object[] provideArgs(Optional input, DataConverter dataConverterWithActivityContext) { - EncodedValues encodedValues = new EncodedValues(input, dataConverterWithActivityContext); - return new Object[] {encodedValues}; + Object[] provideArgs(ActivityInput input, DataConverter dataConverterWithActivityContext) { + return new Object[] {input.getEncodedValues(dataConverterWithActivityContext)}; } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java index 5b2b2bf90a..436fd67c91 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskHandlerImpl.java @@ -103,7 +103,8 @@ public Result handle(ActivityTask activityTask, Scope metricsScope, boolean loca this.namespace, this.taskQueue, localActivity, - activityTask.getCompletionCallback()); + activityTask.getCompletionCallback(), + localActivity ? activityTask.getLocalActivityArguments() : null); ActivityTaskExecutor activity = activities.get(activityType); if (activity != null) { return activity.execute(activityInfo, metricsScope); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityArguments.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityArguments.java new file mode 100644 index 0000000000..0cdcb655ee --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/LocalActivityArguments.java @@ -0,0 +1,82 @@ +package io.temporal.internal.activity; + +import io.temporal.api.common.v1.Payloads; +import io.temporal.common.Copyable; +import io.temporal.common.converter.DataConverter; +import io.temporal.internal.common.CopyableUtils; +import java.lang.reflect.Type; +import java.util.Optional; +import javax.annotation.Nullable; + +/** A scheduling-time snapshot that is never exposed to an activity attempt. */ +public final class LocalActivityArguments { + private final Object[] snapshot; + + private LocalActivityArguments(Object[] snapshot) { + this.snapshot = snapshot; + } + + /** + * Returns a snapshot only when the entire argument list supports copying. Otherwise, the caller + * must use the data converter for the complete list to preserve its batch conversion semantics. + */ + @Nullable + public static LocalActivityArguments capture(@Nullable Object[] arguments) { + if (arguments == null) { + arguments = new Object[0]; + } + for (Object argument : arguments) { + if (argument != null && !(argument instanceof Copyable)) { + return null; + } + } + return new LocalActivityArguments(copy(arguments)); + } + + /** Returns independent arguments for each attempt, including retries after a workflow timer. */ + public Object[] copy() { + return copy(snapshot); + } + + public int size() { + return snapshot.length; + } + + public Object copy(int index) { + return snapshot[index] == null ? null : CopyableUtils.copy(snapshot[index]); + } + + public boolean canCopyTo(Class[] parameterTypes, Type[] genericTypes) { + if (parameterTypes.length != snapshot.length || genericTypes.length != snapshot.length) { + return false; + } + for (int i = 0; i < snapshot.length; i++) { + if (!canCopyTo(i, parameterTypes[i], genericTypes[i])) { + return false; + } + } + return true; + } + + /** Erased generic types and different target classes require the converter's interpretation. */ + public boolean canCopyTo(int index, Class type, Type genericType) { + return index >= 0 + && index < snapshot.length + && type.equals(genericType) + && type.getTypeParameters().length == 0 + && (snapshot[index] == null ? !type.isPrimitive() : snapshot[index].getClass() == type); + } + + /** Preserves conversion behavior for untyped calls with different argument types or counts. */ + public Optional toPayloads(DataConverter converter) { + return converter.toPayloads(snapshot); + } + + private static Object[] copy(Object[] arguments) { + Object[] copy = new Object[arguments.length]; + for (int i = 0; i < arguments.length; i++) { + copy[i] = arguments[i] == null ? null : CopyableUtils.copy(arguments[i]); + } + return copy; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/CopyableUtils.java b/temporal-sdk/src/main/java/io/temporal/internal/common/CopyableUtils.java new file mode 100644 index 0000000000..74aad9bf36 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/CopyableUtils.java @@ -0,0 +1,20 @@ +package io.temporal.internal.common; + +import io.temporal.common.Copyable; +import java.util.Objects; + +/** Validates copies before they cross the local activity boundary. */ +public final class CopyableUtils { + private CopyableUtils() {} + + public static Object copy(Object value) { + Copyable copy = + Objects.requireNonNull( + ((Copyable) value).copy(), "Copyable.copy() must not return null"); + if (!value.getClass().isInstance(copy)) { + throw new IllegalArgumentException( + "Copyable.copy() must return a Copyable compatible with " + value.getClass().getName()); + } + return copy; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/ExecuteLocalActivityParameters.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/ExecuteLocalActivityParameters.java index fed0d132aa..41b8403977 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/ExecuteLocalActivityParameters.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/ExecuteLocalActivityParameters.java @@ -5,6 +5,7 @@ import io.temporal.api.failure.v1.Failure; import io.temporal.api.sdk.v1.UserMetadata; import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.internal.activity.LocalActivityArguments; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.workflow.Functions; import io.temporal.workflow.Workflow; @@ -33,6 +34,7 @@ public class ExecuteLocalActivityParameters { private final @Nullable Duration scheduleToStartTimeout; private @Nullable Functions.Proc onNewAttemptCallback; private final UserMetadata metadata; + private final @Nullable LocalActivityArguments arguments; public ExecuteLocalActivityParameters( @Nonnull PollActivityTaskQueueResponse.Builder activityTaskBuilder, @@ -42,6 +44,26 @@ public ExecuteLocalActivityParameters( boolean doNotIncludeArgumentsIntoMarker, @Nonnull Duration localRetryThreshold, UserMetadata metadata) { + this( + activityTaskBuilder, + scheduleToStartTimeout, + originalScheduledTimestamp, + previousLocalExecutionFailure, + doNotIncludeArgumentsIntoMarker, + localRetryThreshold, + metadata, + null); + } + + public ExecuteLocalActivityParameters( + @Nonnull PollActivityTaskQueueResponse.Builder activityTaskBuilder, + @Nullable Duration scheduleToStartTimeout, + long originalScheduledTimestamp, + @Nullable Failure previousLocalExecutionFailure, + boolean doNotIncludeArgumentsIntoMarker, + @Nonnull Duration localRetryThreshold, + UserMetadata metadata, + @Nullable LocalActivityArguments arguments) { this.activityTaskBuilder = Objects.requireNonNull(activityTaskBuilder, "activityTaskBuilder"); this.scheduleToStartTimeout = scheduleToStartTimeout; this.originalScheduledTimestamp = originalScheduledTimestamp; @@ -50,12 +72,18 @@ public ExecuteLocalActivityParameters( this.localRetryThreshold = localRetryThreshold; this.onNewAttemptCallback = null; this.metadata = metadata; + this.arguments = arguments; } public String getActivityId() { return activityTaskBuilder.getActivityId(); } + @Nullable + public LocalActivityArguments getArguments() { + return arguments; + } + public ActivityType getActivityType() { return activityTaskBuilder.getActivityType(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityCallback.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityCallback.java index a18ae03458..a397a98766 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityCallback.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityCallback.java @@ -11,10 +11,21 @@ @FunctionalInterface public interface LocalActivityCallback extends Functions.Proc2< - Optional, LocalActivityCallback.LocalActivityFailedException> { + LocalActivityCallback.Output, LocalActivityCallback.LocalActivityFailedException> { @Override - void apply(Optional successOutput, LocalActivityFailedException exception); + void apply(Output successOutput, LocalActivityFailedException exception); + + /** Serialized history output and an optional copied value for live execution. */ + final class Output { + public final Optional payloads; + public final @Nullable Object copiedResult; + + public Output(Optional payloads, @Nullable Object copiedResult) { + this.payloads = payloads; + this.copiedResult = copiedResult; + } + } class LocalActivityFailedException extends RuntimeException { private final @Nonnull Failure failure; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityStateMachine.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityStateMachine.java index ab0db39b4e..d7128be635 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityStateMachine.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/LocalActivityStateMachine.java @@ -362,11 +362,15 @@ private void notifyResultFromEvent() { } else { Optional result = Optional.ofNullable(LocalActivityMarkerUtils.getResult(attributes)); - callback.apply(result, null); + callback.apply(new LocalActivityCallback.Output(result, null), null); } } private void notifyResultFromResponse() { - callback.apply(executionSuccess, executionFailure); + callback.apply( + executionFailure == null + ? new LocalActivityCallback.Output(executionSuccess, executionResult.getCopiedResult()) + : null, + executionFailure); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java index 893dc3fe5a..65d764296b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/statemachines/WorkflowStateMachines.java @@ -1318,7 +1318,8 @@ public void handleLocalActivityCompletion(LocalActivityResult laCompletion) { public Functions.Proc scheduleLocalActivityTask( ExecuteLocalActivityParameters parameters, - Functions.Proc2, LocalActivityCallback.LocalActivityFailedException> + Functions.Proc2< + LocalActivityCallback.Output, LocalActivityCallback.LocalActivityFailedException> callback) { checkEventLoopExecuting(); String activityId = parameters.getActivityId(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java index e0b28a77e0..40a3654104 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java @@ -40,6 +40,7 @@ import io.temporal.common.interceptors.WorkflowInboundCallsInterceptor; import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; import io.temporal.failure.*; +import io.temporal.internal.activity.LocalActivityArguments; import io.temporal.internal.common.*; import io.temporal.internal.replay.ChildWorkflowTaskFailedException; import io.temporal.internal.replay.ReplayWorkflowContext; @@ -424,10 +425,11 @@ public void invoke(Optional output, Failure failure) { } private class LocalActivityCallbackImpl implements LocalActivityCallback { - private final CompletablePromise> result = Workflow.newPromise(); + private final CompletablePromise result = Workflow.newPromise(); @Override - public void apply(Optional successOutput, LocalActivityFailedException exception) { + public void apply( + LocalActivityCallback.Output successOutput, LocalActivityFailedException exception) { if (exception != null) { runner.executeInWorkflowThread( "local activity failure callback", () -> result.completeExceptionally(exception)); @@ -450,10 +452,16 @@ public LocalActivityOutput executeLocalActivity(LocalActivityInput inp true); DataConverter dataConverterWithActivityContext = dataConverter.withContext(serializationContext); - Optional payloads = dataConverterWithActivityContext.toPayloads(input.getArgs()); + LocalActivityArguments arguments = LocalActivityArguments.capture(input.getArgs()); + Optional payloads = + arguments == null + ? dataConverterWithActivityContext.toPayloads(input.getArgs()) + : input.getOptions().isDoNotIncludeArgumentsIntoMarker() + ? Optional.empty() + : arguments.toPayloads(dataConverterWithActivityContext); long originalScheduledTime = System.currentTimeMillis(); - CompletablePromise> serializedResult = + CompletablePromise localActivityResult = WorkflowInternal.newCompletablePromise(); executeLocalActivityOverLocalRetryThreshold( input.getActivityName(), @@ -461,22 +469,28 @@ public LocalActivityOutput executeLocalActivity(LocalActivityInput inp input.getOptions(), input.getHeader(), payloads, + arguments, originalScheduledTime, 1, null, - serializedResult); + localActivityResult); // Avoid passing the input to the output handle as it causes the input to be retained for the // duration of the operation. Type resultType = input.getResultType(); Class resultClass = input.getResultClass(); Promise result = - serializedResult.handle( + localActivityResult.handle( (r, f) -> { if (f == null) { - return resultClass != Void.TYPE - ? dataConverterWithActivityContext.fromPayloads(0, r, resultClass, resultType) - : null; + if (resultClass == Void.TYPE) { + return null; + } + if (resultClass.isInstance(r.copiedResult)) { + return resultClass.cast(r.copiedResult); + } + return dataConverterWithActivityContext.fromPayloads( + 0, r.payloads, resultClass, resultType); } else { throw dataConverterWithActivityContext.failureToException( ((LocalActivityCallback.LocalActivityFailedException) f).getFailure()); @@ -486,23 +500,25 @@ public LocalActivityOutput executeLocalActivity(LocalActivityInput inp return new LocalActivityOutput<>(result); } - public void executeLocalActivityOverLocalRetryThreshold( + private void executeLocalActivityOverLocalRetryThreshold( String activityTypeName, @Nullable String activityId, LocalActivityOptions options, Header header, Optional input, + @Nullable LocalActivityArguments arguments, long originalScheduledTime, int attempt, @Nullable Failure previousExecutionFailure, - CompletablePromise> result) { - CompletablePromise> localExecutionResult = + CompletablePromise result) { + CompletablePromise localExecutionResult = executeLocalActivityLocally( activityTypeName, activityId, options, header, input, + arguments, originalScheduledTime, attempt, previousExecutionFailure); @@ -526,6 +542,7 @@ public void executeLocalActivityOverLocalRetryThreshold( options, header, input, + arguments, originalScheduledTime, laException.getLastAttempt() + 1, // Carry the attempt failure, not the local ActivityFailure wrapper. @@ -551,12 +568,13 @@ public void executeLocalActivityOverLocalRetryThreshold( }); } - private CompletablePromise> executeLocalActivityLocally( + private CompletablePromise executeLocalActivityLocally( String activityTypeName, @Nullable String activityId, LocalActivityOptions options, Header header, Optional input, + @Nullable LocalActivityArguments arguments, long originalScheduledTime, int attempt, @Nullable Failure previousExecutionFailure) { @@ -569,6 +587,7 @@ private CompletablePromise> executeLocalActivityLocally( options, header, input, + arguments, attempt, originalScheduledTime, previousExecutionFailure); @@ -654,6 +673,7 @@ private ExecuteLocalActivityParameters constructExecuteLocalActivityParameters( LocalActivityOptions options, Header header, Optional input, + @Nullable LocalActivityArguments arguments, int attempt, long originalScheduledTime, @Nullable Failure previousExecutionFailure) { @@ -709,7 +729,8 @@ private ExecuteLocalActivityParameters constructExecuteLocalActivityParameters( previousExecutionFailure, options.isDoNotIncludeArgumentsIntoMarker(), localRetryThreshold, - userMetadata); + userMetadata, + arguments); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTask.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTask.java index ee2e47362c..bfc3ac3e69 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTask.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTask.java @@ -1,6 +1,7 @@ package io.temporal.internal.worker; import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponseOrBuilder; +import io.temporal.internal.activity.LocalActivityArguments; import io.temporal.worker.tuning.SlotPermit; import io.temporal.workflow.Functions; import javax.annotation.Nonnull; @@ -10,14 +11,24 @@ public final class ActivityTask implements ScalingTask { private final @Nonnull PollActivityTaskQueueResponseOrBuilder response; private final @Nonnull SlotPermit permit; private final @Nonnull Functions.Proc completionCallback; + private final @Nullable LocalActivityArguments localActivityArguments; public ActivityTask( @Nonnull PollActivityTaskQueueResponseOrBuilder response, @Nonnull SlotPermit permit, @Nonnull Functions.Proc completionCallback) { + this(response, permit, completionCallback, null); + } + + public ActivityTask( + @Nonnull PollActivityTaskQueueResponseOrBuilder response, + @Nonnull SlotPermit permit, + @Nonnull Functions.Proc completionCallback, + @Nullable LocalActivityArguments localActivityArguments) { this.response = response; this.permit = permit; this.completionCallback = completionCallback; + this.localActivityArguments = localActivityArguments; } @Nonnull @@ -25,6 +36,11 @@ public PollActivityTaskQueueResponseOrBuilder getResponse() { return response; } + @Nullable + public LocalActivityArguments getLocalActivityArguments() { + return localActivityArguments; + } + /** * Completion handle function that must be called by the handler whenever activity processing is * completed. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTaskHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTaskHandler.java index d1dd702668..a98c17e4c9 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTaskHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityTaskHandler.java @@ -19,6 +19,7 @@ final class Result { private final TaskFailedResult taskFailed; private final RespondActivityTaskCanceledRequest taskCanceled; private final boolean manualCompletion; + private final Object copiedResult; @Override public String toString() { @@ -65,11 +66,22 @@ public Result( TaskFailedResult taskFailed, RespondActivityTaskCanceledRequest taskCanceled, boolean manualCompletion) { + this(activityId, taskCompleted, taskFailed, taskCanceled, manualCompletion, null); + } + + public Result( + String activityId, + RespondActivityTaskCompletedRequest taskCompleted, + TaskFailedResult taskFailed, + RespondActivityTaskCanceledRequest taskCanceled, + boolean manualCompletion, + Object copiedResult) { this.activityId = activityId; this.taskCompleted = taskCompleted; this.taskFailed = taskFailed; this.taskCanceled = taskCanceled; this.manualCompletion = manualCompletion; + this.copiedResult = copiedResult; } public String getActivityId() { @@ -80,6 +92,11 @@ public RespondActivityTaskCompletedRequest getTaskCompleted() { return taskCompleted; } + /** A copied local activity result for live execution, or null if unavailable. */ + public Object getCopiedResult() { + return copiedResult; + } + public TaskFailedResult getTaskFailed() { return taskFailed; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityExecutionContext.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityExecutionContext.java index a93e916fbd..95eebfc728 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityExecutionContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityExecutionContext.java @@ -5,6 +5,7 @@ import io.temporal.api.workflow.v1.PendingActivityInfo; import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest; import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.internal.activity.LocalActivityArguments; import io.temporal.internal.statemachines.ExecuteLocalActivityParameters; import io.temporal.worker.tuning.SlotPermit; import io.temporal.workflow.Functions; @@ -58,6 +59,11 @@ public String getActivityId() { return executionParams.getActivityId(); } + @Nullable + public LocalActivityArguments getArguments() { + return executionParams.getArguments(); + } + public int getCurrentAttempt() { return currentAttempt.get(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityResult.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityResult.java index d81779d439..9abe7e2a6a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityResult.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityResult.java @@ -14,6 +14,7 @@ public final class LocalActivityResult { private final @Nullable RespondActivityTaskCompletedRequest executionCompleted; private final @Nullable ExecutionFailedResult executionFailed; private final @Nullable RespondActivityTaskCanceledRequest executionCanceled; + private final @Nullable Object copiedResult; /** * If present, it will cause an immediate WFT failure instead of providing LA result to the @@ -23,7 +24,13 @@ public final class LocalActivityResult { static LocalActivityResult completed(ActivityTaskHandler.Result ahResult, int attempt) { return new LocalActivityResult( - ahResult.getActivityId(), attempt, ahResult.getTaskCompleted(), null, null, null); + ahResult.getActivityId(), + attempt, + ahResult.getTaskCompleted(), + null, + null, + null, + ahResult.getCopiedResult()); } static LocalActivityResult failed( @@ -60,12 +67,31 @@ public LocalActivityResult( @Nullable ExecutionFailedResult executionFailed, @Nullable RespondActivityTaskCanceledRequest executionCanceled, @Nullable ProcessingErrorResult processingError) { + this( + activityId, + lastAttempt, + executionCompleted, + executionFailed, + executionCanceled, + processingError, + null); + } + + public LocalActivityResult( + @Nonnull String activityId, + int lastAttempt, + @Nullable RespondActivityTaskCompletedRequest executionCompleted, + @Nullable ExecutionFailedResult executionFailed, + @Nullable RespondActivityTaskCanceledRequest executionCanceled, + @Nullable ProcessingErrorResult processingError, + @Nullable Object copiedResult) { this.activityId = activityId; this.lastAttempt = lastAttempt; this.executionCompleted = executionCompleted; this.executionFailed = executionFailed; this.executionCanceled = executionCanceled; this.processingError = processingError; + this.copiedResult = copiedResult; } @Nonnull @@ -82,6 +108,11 @@ public RespondActivityTaskCompletedRequest getExecutionCompleted() { return executionCompleted; } + @Nullable + public Object getCopiedResult() { + return copiedResult; + } + @Nullable public ExecutionFailedResult getExecutionFailed() { return executionFailed; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityWorker.java index d59b351bec..0d59109dc6 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityWorker.java @@ -454,7 +454,11 @@ public void handle(LocalActivityAttemptTask attemptTask) throws Exception { try { activityHandlerResult = handler.handle( - new ActivityTask(activityTask, executionContext.getPermit(), () -> {}), + new ActivityTask( + activityTask, + executionContext.getPermit(), + () -> {}, + executionContext.getArguments()), metricsScope, true); } finally { diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityInfoImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityInfoImplTest.java index 41f20a634e..2e055b1da0 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityInfoImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityInfoImplTest.java @@ -35,7 +35,8 @@ public void testWorkflowActivityReturnsWorkflowInfo() { io.temporal.api.common.v1.WorkflowType.newBuilder().setName("MyWorkflow")) .build(); - ActivityInfoImpl info = new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null); + ActivityInfoImpl info = + new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null, null); assertEquals("wf-id", info.getWorkflowId()); assertEquals("run-id", info.getWorkflowRunId()); assertEquals("MyWorkflow", info.getWorkflowType()); @@ -47,7 +48,8 @@ public void testStandaloneActivityReturnsNullWorkflowInfo() { // No workflowExecution set PollActivityTaskQueueResponse response = baseResponse().build(); - ActivityInfoImpl info = new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null); + ActivityInfoImpl info = + new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null, null); assertNull(info.getWorkflowId()); assertNull(info.getWorkflowRunId()); assertNull(info.getWorkflowType()); @@ -62,7 +64,8 @@ public void testEmptyWorkflowIdTreatedAsStandalone() { WorkflowExecution.newBuilder().setWorkflowId("").setRunId("").build()) .build(); - ActivityInfoImpl info = new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null); + ActivityInfoImpl info = + new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null, null); assertNull(info.getWorkflowId()); assertNull(info.getWorkflowRunId()); assertFalse(info.isInWorkflow()); @@ -71,7 +74,8 @@ public void testEmptyWorkflowIdTreatedAsStandalone() { @Test public void testEmptyActivityRunIdTreatedAsNull() { PollActivityTaskQueueResponse response = baseResponse().setActivityRunId("").build(); - ActivityInfoImpl info = new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null); + ActivityInfoImpl info = + new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null, null); assertNull(info.getActivityRunId()); } @@ -84,7 +88,8 @@ public void testEmptyWorkflowTypeTreatedAsNull() { .setWorkflowType(io.temporal.api.common.v1.WorkflowType.newBuilder().setName("")) .build(); - ActivityInfoImpl info = new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null); + ActivityInfoImpl info = + new ActivityInfoImpl(response, NAMESPACE, "task-queue", false, null, null); // workflowId is set, but workflowType is empty → getWorkflowType() returns null assertEquals("wf-id", info.getWorkflowId()); assertNull(info.getWorkflowType()); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/LocalActivityArgumentsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/LocalActivityArgumentsTest.java new file mode 100644 index 0000000000..0d4d3a234f --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/LocalActivityArgumentsTest.java @@ -0,0 +1,111 @@ +package io.temporal.internal.activity; + +import static org.junit.Assert.*; +import static org.mockito.Mockito.*; + +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.common.Copyable; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.common.converter.EncodedValues; +import io.temporal.internal.common.CopyableUtils; +import org.junit.Test; + +public class LocalActivityArgumentsTest { + private static class Value implements Copyable { + int value; + + @Override + public Value copy() { + Value copy = new Value(); + copy.value = value; + return copy; + } + } + + @Test + public void capturesSnapshotAndCopiesEachAttemptIncludingNullSlots() { + Value original = new Value(); + Object[] input = {original, null}; + LocalActivityArguments snapshot = LocalActivityArguments.capture(input); + original.value = 1; + input[0] = null; + Object[] first = snapshot.copy(); + ((Value) first[0]).value = 2; + first[1] = original; + Object[] second = snapshot.copy(); + assertNotSame(first[0], second[0]); + assertEquals(0, ((Value) second[0]).value); + assertNull(second[1]); + Class[] compatible = {Value.class, Object.class}; + Class[] incompatible = {String.class, Object.class}; + Class[] primitive = {Value.class, int.class}; + assertTrue(snapshot.canCopyTo(compatible, compatible)); + assertFalse(snapshot.canCopyTo(incompatible, incompatible)); + assertFalse(snapshot.canCopyTo(primitive, primitive)); + assertFalse(snapshot.canCopyTo(new Class[0], new Class[0])); + } + + @Test + public void activityInputSelectsOneArgumentSource() { + DataConverter converter = DefaultDataConverter.STANDARD_INSTANCE; + PollActivityTaskQueueResponse response = + PollActivityTaskQueueResponse.newBuilder() + .setInput(converter.toPayloads("serialized").get()) + .build(); + Value original = new Value(); + ActivityInput copied = + new ActivityInput( + PollActivityTaskQueueResponse.getDefaultInstance(), + LocalActivityArguments.capture(new Object[] {original})); + DataConverter unusedConverter = mock(DataConverter.class); + Class[] copyTypes = {Value.class}; + assertNotSame(original, copied.getArguments(copyTypes, copyTypes, unusedConverter)[0]); + verifyNoInteractions(unusedConverter); + + Class[] serializedTypes = {String.class}; + assertArrayEquals( + new Object[] {"serialized"}, + new ActivityInput(response, null) + .getArguments(serializedTypes, serializedTypes, converter)); + } + + @Test + public void emptyDynamicInputUsesConverterDefaults() { + EncodedValues values = + new ActivityInput( + PollActivityTaskQueueResponse.getDefaultInstance(), + LocalActivityArguments.capture(new Object[0])) + .getEncodedValues(DefaultDataConverter.STANDARD_INSTANCE); + assertEquals(0, values.getSize()); + assertNull(values.get(0, String.class)); + assertEquals(Integer.valueOf(0), values.get(0, int.class)); + } + + @Test + public void immutableCopiesMayReturnThisButMixedBatchesMustNotCopy() { + Value immutable = + new Value() { + @Override + public Value copy() { + return this; + } + }; + assertSame(immutable, LocalActivityArguments.capture(new Object[] {immutable}).copy()[0]); + Copyable mustNotCopy = + () -> { + throw new AssertionError("Mixed batches must use serialization."); + }; + assertNull(LocalActivityArguments.capture(new Object[] {mustNotCopy, "ordinary"})); + } + + @Test(expected = NullPointerException.class) + public void rejectsNullCopies() { + CopyableUtils.copy((Copyable) () -> null); + } + + @Test(expected = IllegalArgumentException.class) + public void rejectsCopiesOfAnotherType() { + CopyableUtils.copy((Copyable) Value::new); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/LocalActivityStateMachineTest.java b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/LocalActivityStateMachineTest.java index 6329d5b8cb..469827bd0b 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/statemachines/LocalActivityStateMachineTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/statemachines/LocalActivityStateMachineTest.java @@ -108,16 +108,16 @@ protected void buildWorkflow(AsyncWorkflowBuilder builder) { null); builder - ., LocalActivityCallback.LocalActivityFailedException>add2( + .add2( (r, c) -> stateMachines.scheduleLocalActivityTask(parameters1, c)) .add((r) -> stateMachines.completeWorkflow(Optional.empty())); builder - ., LocalActivityCallback.LocalActivityFailedException>add2( + .add2( (r, c) -> stateMachines.scheduleLocalActivityTask(parameters2, c)) - ., LocalActivityCallback.LocalActivityFailedException>add2( + .add2( (r, c) -> stateMachines.scheduleLocalActivityTask(parameters3, c)) - .add((r) -> result = r.getT1()); + .add((r) -> result = r.getT1().payloads); } } /* @@ -289,7 +289,7 @@ protected void buildWorkflow(AsyncWorkflowBuilder builder) { null, null); builder - ., LocalActivityCallback.LocalActivityFailedException>add2( + .add2( (r, c) -> stateMachines.scheduleLocalActivityTask(parameters1, c)) .add((r) -> stateMachines.completeWorkflow(Optional.empty())); } @@ -375,7 +375,7 @@ protected void buildWorkflow(AsyncWorkflowBuilder builder) { (r, c) -> { completionCallback.set(c); }) - ., LocalActivityCallback.LocalActivityFailedException>add2( + .add2( (r, c) -> stateMachines.scheduleLocalActivityTask(parameters1, c)); } } diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/LocalActivityCopyableTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/LocalActivityCopyableTest.java new file mode 100644 index 0000000000..008211d9c6 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/LocalActivityCopyableTest.java @@ -0,0 +1,411 @@ +package io.temporal.workflow.activityTests; + +import static org.junit.Assert.*; + +import com.google.common.reflect.TypeToken; +import io.temporal.activity.*; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.enums.v1.EventType; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.common.Copyable; +import io.temporal.common.RetryOptions; +import io.temporal.common.converter.*; +import io.temporal.failure.ApplicationFailure; +import io.temporal.testing.*; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.*; +import java.lang.reflect.Type; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +@RunWith(Parameterized.class) +public class LocalActivityCopyableTest { + @Parameterized.Parameters(name = "includeMarkerArguments={0}") + public static Object[] markerSettings() { + return new Object[] {false, true}; + } + + @Parameterized.Parameter public boolean includeMarkerArguments; + + private final CountingConverter converter = new CountingConverter(); + private final CopyActivitiesImpl activities = new CopyActivitiesImpl(); + private final WorkflowClientOptions clientOptions = + WorkflowClientOptions.newBuilder() + .setDataConverter( + DefaultDataConverter.newDefaultInstance().withPayloadConverterOverrides(converter)) + .build(); + + @Rule + public SDKTestWorkflowRule rule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowClientOptions(clientOptions) + .setWorkflowTypes(CopyWorkflowImpl.class) + .setActivityImplementations( + activities, (DynamicActivity) LocalActivityCopyableTest::executeDynamic) + .build(); + + private static int executeDynamic(EncodedValues args) { + assertEquals(1, args.getSize()); + if (Activity.getExecutionContext().getInfo().getActivityType().equals("DynamicGeneric")) { + Box converted = args.get(0, Box.class, new TypeToken>() {}.getType()); + assertEquals(Long.class, converted.value.getClass()); + assertEquals(Long.valueOf(1), converted.value); + return 1; + } + MutableValue input = args.get(0, MutableValue.class); + int original = input.value; + input.value = 100; + MutableValue second = args.get(0, MutableValue.class, MutableValue.class); + assertNotSame(input, second); + assertEquals(original, second.value); + second.value = 200; + assertEquals(original, args.get(0, MutableValue.class).value); + if (input.mode.equals("dynamicFallback")) { + assertEquals(original, args.get(0, PlainValue.class).value); + assertNull(args.get(1, MutableValue.class)); + assertEquals(0, (int) args.get(1, int.class)); + assertEquals(original, args.get(0, PlainValue.class, PlainValue.class).value); + EncodedValues snapshot = + new EncodedValues(args.toPayloads(), DefaultDataConverter.newDefaultInstance()); + assertEquals(original, snapshot.get(0, MutableValue.class).value); + assertEquals(args.toPayloads(), snapshot.toPayloads()); + assertEquals(1, args.getSize()); + } + return original; + } + + @Test + public void schedulingSnapshotAndActivityMutationAreIsolated() { + assertEquals("1:2", run("snapshot")); + assertEquals(Collections.singletonList(1), activities.inputs); + assertConversions(includeMarkerArguments ? 1 : 0, 0); + } + + @Test + public void eachLocalRetryReceivesAFreshSnapshot() throws Exception { + assertRetries("localRetry", false); + } + + @Test + public void eachTimerBackedRetryReceivesAFreshSnapshot() throws Exception { + assertRetries("timerRetry", true); + } + + private void assertRetries(String mode, boolean timerBacked) throws Exception { + CopyWorkflow workflow = rule.newWorkflowStub(CopyWorkflow.class); + assertEquals("1:1", workflow.execute(mode, includeMarkerArguments)); + assertEquals(Arrays.asList(1, 1, 1), activities.inputs); + assertNotSame(activities.attemptInputs.get(0), activities.attemptInputs.get(1)); + assertNotSame(activities.attemptInputs.get(1), activities.attemptInputs.get(2)); + assertConversions(includeMarkerArguments ? 1 : 0, 0); + assertEquals( + timerBacked ? 2 : 0, + rule.getHistoryEvents( + WorkflowStub.fromTyped(workflow).getExecution().getWorkflowId(), + EventType.EVENT_TYPE_TIMER_STARTED) + .size()); + replay(workflow); + assertConversions(includeMarkerArguments ? 2 : 0, 0); + assertEquals(3, activities.attemptInputs.size()); + } + + @Test + public void outputMutationIsIsolatedAndReplayDeserializesTheSnapshot() throws Exception { + CopyWorkflow workflow = rule.newWorkflowStub(CopyWorkflow.class); + assertEquals("7:7:8", workflow.execute("output", includeMarkerArguments)); + assertConversions(1, 0); + assertEquals(9, activities.retained.value); + replay(workflow); + assertConversions(1, 1); + assertEquals(1, activities.outputCalls.get()); + } + + @Test + public void mixedArgumentsUseFullSerialization() { + assertEquals("1:1", run("mixed")); + assertConversions(1, 1); + } + + @Test + public void regularActivitiesAndSideEffectsStillUseSerialization() { + assertEquals("1:6", run("regular")); + assertConversions(2, 2); + } + + @Test + public void dynamicInputAndIncompatibleUntypedOutputAreSupported() { + assertEquals("1:1", run("dynamic")); + assertConversions(includeMarkerArguments ? 1 : 0, 0); + assertEquals("7", run("untypedOutput")); + assertConversions(includeMarkerArguments ? 2 : 1, 0); + assertEquals(1, converter.plainReads.get()); + } + + @Test + public void dynamicFallbackUsesTheSchedulingSnapshotAndMissingArgumentDefaults() { + assertEquals("1:1", run("dynamicFallback")); + assertConversions(1, 0); + assertEquals(2, converter.plainReads.get()); + } + + @Test + public void dynamicGenericInputHonorsTheRequestedType() { + assertEquals("1", run("dynamicGeneric")); + } + + @Test + public void pojoGenericInputHonorsTheDeclaredType() { + assertEquals("1", run("pojoGeneric")); + } + + private String run(String mode) { + return rule.newWorkflowStub(CopyWorkflow.class).execute(mode, includeMarkerArguments); + } + + private void assertConversions(int writes, int reads) { + assertEquals("MutableValue serialization count", writes, converter.writes.get()); + assertEquals("MutableValue deserialization count", reads, converter.reads.get()); + } + + private void replay(CopyWorkflow workflow) throws Exception { + try (TestWorkflowEnvironment replay = + TestWorkflowEnvironment.newInstance( + TestEnvironmentOptions.newBuilder().setWorkflowClientOptions(clientOptions).build())) { + WorkflowReplayer.replayWorkflowExecution( + rule.getWorkflowClient() + .fetchHistory(WorkflowStub.fromTyped(workflow).getExecution().getWorkflowId()), + replay, + CopyWorkflowImpl.class); + } + } + + public static class MutableValue implements Copyable { + public int value; + public String mode; + + public MutableValue() {} + + MutableValue(int value, String mode) { + this.value = value; + this.mode = mode; + } + + @Override + public MutableValue copy() { + return new MutableValue(value, mode); + } + } + + public static class PlainValue { + public int value; + public String mode; + } + + public static class Box implements Copyable> { + public T value; + + public Box() {} + + Box(T value) { + this.value = value; + } + + @Override + public Box copy() { + return new Box<>(value); + } + } + + @WorkflowInterface + public interface CopyWorkflow { + @WorkflowMethod + String execute(String mode, boolean includeMarkerArguments); + } + + public static class CopyWorkflowImpl implements CopyWorkflow { + @Override + public String execute(String mode, boolean includeMarkerArguments) { + boolean timerRetry = mode.equals("timerRetry"); + LocalActivityOptions options = + LocalActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .setDoNotIncludeArgumentsIntoMarker(!includeMarkerArguments) + .setLocalRetryThreshold(timerRetry ? Duration.ofMillis(1) : Duration.ofSeconds(5)) + .setRetryOptions( + RetryOptions.newBuilder() + .setInitialInterval( + timerRetry ? Duration.ofSeconds(1) : Duration.ofMillis(10)) + .setBackoffCoefficient(1) + .setMaximumAttempts(3) + .build()) + .build(); + CopyActivities local = Workflow.newLocalActivityStub(CopyActivities.class, options); + MutableValue input = new MutableValue(1, mode); + switch (mode) { + case "snapshot": + boolean[] started = {false}; + Promise result = + Async.function( + () -> { + started[0] = true; + return local.mutate(input); + }); + // The async thread yields only after scheduling the activity. + Workflow.await(() -> started[0]); + input.value = 2; + local.release(); + return result.get() + ":" + input.value; + case "localRetry": + case "timerRetry": + return local.mutate(input) + ":" + input.value; + case "output": + MutableValue output = local.output(); + int original = output.value; + output.value = 8; + return original + ":" + local.changeRetained() + ":" + output.value; + case "mixed": + return local.mixed(input, "ordinary") + ":" + input.value; + case "regular": + CopyActivities regular = + Workflow.newActivityStub( + CopyActivities.class, + ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build()); + int value = regular.mutate(input); + MutableValue sideEffect = + Workflow.sideEffect(MutableValue.class, () -> new MutableValue(6, mode)); + return value + ":" + sideEffect.value; + case "dynamicGeneric": + return Integer.toString( + Workflow.newUntypedLocalActivityStub(options) + .execute("DynamicGeneric", Integer.class, new Box<>(1))); + case "pojoGeneric": + return Long.toString( + Workflow.newUntypedLocalActivityStub(options) + .execute("ReadBox", Long.class, new Box<>(1))); + case "dynamic": + case "dynamicFallback": + return Workflow.newUntypedLocalActivityStub(options) + .execute("DynamicCopy", Integer.class, input) + + ":" + + input.value; + case "untypedOutput": + PlainValue converted = + Workflow.newUntypedLocalActivityStub(options).execute("CopyOutput", PlainValue.class); + return Integer.toString(converted.value); + default: + throw new IllegalArgumentException(mode); + } + } + } + + @ActivityInterface + public interface CopyActivities { + int mutate(MutableValue value); + + void release(); + + @ActivityMethod(name = "CopyOutput") + MutableValue output(); + + int changeRetained(); + + int mixed(MutableValue copyable, String ordinary); + + @ActivityMethod(name = "ReadBox") + long readBox(Box value); + } + + public static class CopyActivitiesImpl implements CopyActivities { + private final CountDownLatch gate = new CountDownLatch(1); + final List inputs = Collections.synchronizedList(new ArrayList<>()); + final List attemptInputs = Collections.synchronizedList(new ArrayList<>()); + final AtomicInteger outputCalls = new AtomicInteger(); + private volatile MutableValue retained; + + @Override + public int mutate(MutableValue value) { + if (value.mode.equals("snapshot")) { + try { + if (!gate.await(5, TimeUnit.SECONDS)) { + throw new AssertionError("Workflow did not release the activity."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + int original = value.value; + inputs.add(original); + attemptInputs.add(value); + value.value = 100; + if (value.mode.endsWith("Retry") + && Activity.getExecutionContext().getInfo().getAttempt() < 3) { + throw ApplicationFailure.newFailure("Retry after mutating the input.", "retry"); + } + return original; + } + + @Override + public void release() { + gate.countDown(); + } + + @Override + public MutableValue output() { + outputCalls.incrementAndGet(); + retained = new MutableValue(7, "output"); + return retained; + } + + @Override + public int changeRetained() { + int original = retained.value; + retained.value = 9; + return original; + } + + @Override + public int mixed(MutableValue copyable, String ordinary) { + return mutate(copyable); + } + + @Override + public long readBox(Box value) { + return value.value.longValue(); + } + } + + private static class CountingConverter extends JacksonJsonPayloadConverter { + final AtomicInteger writes = new AtomicInteger(); + final AtomicInteger reads = new AtomicInteger(); + final AtomicInteger plainReads = new AtomicInteger(); + + @Override + public Optional toData(Object value) { + if (value instanceof MutableValue) { + writes.incrementAndGet(); + } + return super.toData(value); + } + + @Override + public T fromData(Payload content, Class valueClass, Type valueType) { + if (valueClass == MutableValue.class) { + reads.incrementAndGet(); + } else if (valueClass == PlainValue.class) { + plainReads.incrementAndGet(); + } + return super.fromData(content, valueClass, valueType); + } + } +}