diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java index 357df1c4da..3b3d7f6f7d 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java @@ -41,20 +41,37 @@ public interface WorkflowOutboundCallsInterceptor { final class ActivityInput { private final String activityName; + private final @Nullable String activityId; private final Class resultClass; private final Type resultType; private final Object[] args; private final ActivityOptions options; private final Header header; + /** + * @deprecated Kept only for backward compatibility. + */ + @Deprecated + public ActivityInput( + String activityName, + Class resultClass, + Type resultType, + Object[] args, + ActivityOptions options, + Header header) { + this(activityName, null, resultClass, resultType, args, options, header); + } + public ActivityInput( String activityName, + @Nullable String activityId, Class resultClass, Type resultType, Object[] args, ActivityOptions options, Header header) { this.activityName = activityName; + this.activityId = activityId; this.resultClass = resultClass; this.resultType = resultType; this.args = args; @@ -66,6 +83,12 @@ public String getActivityName() { return activityName; } + /** Returns the caller-supplied Activity ID, or {@code null} if the SDK should generate one. */ + @Nullable + public String getActivityId() { + return activityId; + } + public Class getResultClass() { return resultClass; } @@ -107,20 +130,37 @@ public Promise getResult() { final class LocalActivityInput { private final String activityName; + private final @Nullable String activityId; private final Class resultClass; private final Type resultType; private final Object[] args; private final LocalActivityOptions options; private final Header header; + /** + * @deprecated Kept only for backward compatibility. + */ + @Deprecated + public LocalActivityInput( + String activityName, + Class resultClass, + Type resultType, + Object[] args, + LocalActivityOptions options, + Header header) { + this(activityName, null, resultClass, resultType, args, options, header); + } + public LocalActivityInput( String activityName, + @Nullable String activityId, Class resultClass, Type resultType, Object[] args, LocalActivityOptions options, Header header) { this.activityName = activityName; + this.activityId = activityId; this.resultClass = resultClass; this.resultType = resultType; this.args = args; @@ -132,6 +172,12 @@ public String getActivityName() { return activityName; } + /** Returns the caller-supplied Activity ID, or {@code null} if the SDK should generate one. */ + @Nullable + public String getActivityId() { + return activityId; + } + public Class getResultClass() { return resultClass; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationHandler.java index e46408ca06..e0635261e4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationHandler.java @@ -4,6 +4,7 @@ import io.temporal.activity.ActivityOptions; import io.temporal.common.MethodRetry; import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; +import io.temporal.workflow.ActivityInvocationOptions; import io.temporal.workflow.ActivityStub; import io.temporal.workflow.Functions; import java.lang.reflect.InvocationHandler; @@ -46,7 +47,6 @@ private ActivityInvocationHandler( @Override protected Function getActivityFunc( Method method, MethodRetry methodRetry, String activityName) { - Function function; ActivityOptions merged = ActivityOptions.newBuilder(options) .mergeActivityOptions(this.activityMethodOptions.get(activityName)) @@ -58,10 +58,15 @@ protected Function getActivityFunc( + activityName + " activity. Please set at least one of the above through the ActivityStub or WorkflowImplementationOptions."); } + ActivityInvocationOptions invocationOptions = ActivityInvocationInternal.consumeOptions(); ActivityStub stub = ActivityStubImpl.newInstance(merged, activityExecutor, assertReadOnly); - function = - (a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a); - return function; + return (a) -> + stub.execute( + activityName, + method.getReturnType(), + method.getGenericReturnType(), + invocationOptions, + a); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationInternal.java new file mode 100644 index 0000000000..42a0b43781 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationInternal.java @@ -0,0 +1,118 @@ +package io.temporal.internal.sync; + +import io.temporal.workflow.ActivityInvocationOptions; +import io.temporal.workflow.Functions; +import io.temporal.workflow.Promise; +import java.util.Objects; + +/** Applies options to one typed Activity proxy invocation. */ +final class ActivityInvocationInternal { + + private static final ThreadLocal invocation = new ThreadLocal<>(); + private static final ActivityInvocationOptions DEFAULT_OPTIONS = + ActivityInvocationOptions.newBuilder().build(); + + private ActivityInvocationInternal() {} + + static ActivityInvocationOptions getDefaultOptions() { + return DEFAULT_OPTIONS; + } + + static R invoke(ActivityInvocationOptions options, Functions.Func invocationFunction) { + State state = startInvocation(options, false); + try { + R result = invocationFunction.apply(); + state.verifyConsumed(); + return result; + } finally { + invocation.remove(); + } + } + + static void invoke(ActivityInvocationOptions options, Functions.Proc invocationFunction) { + State state = startInvocation(options, false); + try { + invocationFunction.apply(); + state.verifyConsumed(); + } finally { + invocation.remove(); + } + } + + static Promise invokeAsync( + ActivityInvocationOptions options, Functions.Proc invocationFunction) { + State state = startInvocation(options, true); + try { + invocationFunction.apply(); + return state.getResult(); + } finally { + invocation.remove(); + } + } + + private static State startInvocation(ActivityInvocationOptions options, boolean async) { + if (invocation.get() != null) { + throw new IllegalStateException("Already invoking an Activity with invocation options"); + } + + State state = new State(Objects.requireNonNull(options, "options"), async); + invocation.set(state); + return state; + } + + static ActivityInvocationOptions consumeOptions() { + State state = invocation.get(); + if (state == null) { + return DEFAULT_OPTIONS; + } + if (state.consumed) { + throw new IllegalStateException("ActivityInvocationOptions can apply to only one invocation"); + } + state.consumed = true; + return state.options; + } + + static boolean captureResult(Promise result) { + State state = invocation.get(); + if (state == null || !state.async) { + return false; + } + if (state.result != null) { + throw new IllegalStateException("ActivityInvocationOptions can apply to only one invocation"); + } + state.result = Objects.requireNonNull(result, "result"); + return true; + } + + private static final class State { + private final ActivityInvocationOptions options; + private final boolean async; + private boolean consumed; + private Promise result; + + private State(ActivityInvocationOptions options, boolean async) { + this.options = options; + this.async = async; + } + + private void verifyConsumed() { + if (!consumed) { + throw invalidInvocation(); + } + } + + @SuppressWarnings("unchecked") + private Promise getResult() { + if (!consumed || result == null) { + throw invalidInvocation(); + } + return (Promise) result; + } + + private IllegalArgumentException invalidInvocation() { + return new IllegalArgumentException( + "activityMethod must invoke an Activity stub created through Workflow.newActivityStub " + + "or Workflow.newLocalActivityStub"); + } + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubBase.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubBase.java index 95698f6ef8..755949148a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubBase.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubBase.java @@ -2,11 +2,12 @@ import com.google.common.base.Defaults; import io.temporal.failure.ActivityFailure; +import io.temporal.workflow.ActivityInvocationOptions; import io.temporal.workflow.ActivityStub; import io.temporal.workflow.Promise; import java.lang.reflect.Type; -/** Supports calling activity by name and arguments without its strongly typed interface. */ +/** Supports calling an activity by name and arguments without its strongly typed interface. */ abstract class ActivityStubBase implements ActivityStub { @Override @@ -16,7 +17,46 @@ public T execute(String activityName, Class resultClass, Object... args) @Override public T execute(String activityName, Class resultClass, Type resultType, Object... args) { - Promise result = executeAsync(activityName, resultClass, resultType, args); + return getResult(executeAsync(activityName, resultClass, resultType, args), resultClass); + } + + @Override + public Promise executeAsync(String activityName, Class resultClass, Object... args) { + return executeAsync(activityName, resultClass, resultClass, args); + } + + @Override + public Promise executeAsync( + String activityName, Class resultClass, Type resultType, Object... args) { + return executeAsync( + activityName, + resultClass, + resultType, + ActivityInvocationInternal.getDefaultOptions(), + args); + } + + @Override + public R execute( + String activityName, + Class resultClass, + ActivityInvocationOptions options, + Object... args) { + return execute(activityName, resultClass, resultClass, options, args); + } + + @Override + public abstract R execute( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions options, + Object... args); + + protected R getResult(Promise result, Class resultClass) { + if (ActivityInvocationInternal.captureResult(result)) { + return Defaults.defaultValue(resultClass); + } if (AsyncInternal.isAsync()) { AsyncInternal.setAsyncResult(result); return Defaults.defaultValue(resultClass); @@ -33,11 +73,19 @@ public T execute(String activityName, Class resultClass, Type resultType, } @Override - public Promise executeAsync(String activityName, Class resultClass, Object... args) { - return executeAsync(activityName, resultClass, resultClass, args); + public Promise executeAsync( + String activityName, + Class resultClass, + ActivityInvocationOptions options, + Object... args) { + return executeAsync(activityName, resultClass, resultClass, options, args); } @Override public abstract Promise executeAsync( - String activityName, Class resultClass, Type resultType, Object... args); + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions options, + Object... args); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubImpl.java index 8ed9e62be3..47610c74e9 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubImpl.java @@ -3,13 +3,15 @@ import io.temporal.activity.ActivityOptions; import io.temporal.common.interceptors.Header; import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; +import io.temporal.workflow.ActivityInvocationOptions; import io.temporal.workflow.ActivityStub; import io.temporal.workflow.Functions; import io.temporal.workflow.Promise; import java.lang.reflect.Type; +import java.util.Objects; final class ActivityStubImpl extends ActivityStubBase { - protected final ActivityOptions options; + private final ActivityOptions options; private final WorkflowOutboundCallsInterceptor activityExecutor; private final Functions.Proc assertReadOnly; @@ -31,14 +33,49 @@ static ActivityStub newInstance( this.assertReadOnly = assertReadOnly; } + @Override + public R execute( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions invocationOptions, + Object... args) { + Objects.requireNonNull(invocationOptions, "invocationOptions"); + return getResult( + scheduleActivity( + activityName, resultClass, resultType, invocationOptions.getActivityId(), args), + resultClass); + } + @Override public Promise executeAsync( - String activityName, Class resultClass, Type resultType, Object... args) { + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions invocationOptions, + Object... args) { + Objects.requireNonNull(invocationOptions, "invocationOptions"); + return scheduleActivity( + activityName, resultClass, resultType, invocationOptions.getActivityId(), args); + } + + private Promise scheduleActivity( + String activityName, + Class resultClass, + Type resultType, + String activityId, + Object... args) { this.assertReadOnly.apply(); return activityExecutor .executeActivity( new WorkflowOutboundCallsInterceptor.ActivityInput<>( - activityName, resultClass, resultType, args, options, Header.empty())) + activityName, + activityId, + resultClass, + resultType, + args, + this.options, + Header.empty())) .getResult(); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityInvocationHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityInvocationHandler.java index 5b173d33f3..e147104064 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityInvocationHandler.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityInvocationHandler.java @@ -4,6 +4,7 @@ import io.temporal.activity.LocalActivityOptions; import io.temporal.common.MethodRetry; import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; +import io.temporal.workflow.ActivityInvocationOptions; import io.temporal.workflow.ActivityStub; import io.temporal.workflow.Functions; import java.lang.reflect.InvocationHandler; @@ -47,7 +48,6 @@ private LocalActivityInvocationHandler( @Override public Function getActivityFunc( Method method, MethodRetry methodRetry, String activityName) { - Function function; LocalActivityOptions mergedOptions = LocalActivityOptions.newBuilder(options) .mergeActivityOptions(activityMethodOptions.get(activityName)) @@ -55,9 +55,14 @@ public Function getActivityFunc( .build(); ActivityStub stub = LocalActivityStubImpl.newInstance(mergedOptions, activityExecutor, assertReadOnly); - function = - (a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a); - return function; + ActivityInvocationOptions invocationOptions = ActivityInvocationInternal.consumeOptions(); + return (a) -> + stub.execute( + activityName, + method.getReturnType(), + method.getGenericReturnType(), + invocationOptions, + a); } @Override diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityStubImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityStubImpl.java index 6744c26cde..191fee0b09 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityStubImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityStubImpl.java @@ -3,10 +3,13 @@ import io.temporal.activity.LocalActivityOptions; import io.temporal.common.interceptors.Header; import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor; +import io.temporal.workflow.ActivityInvocationOptions; import io.temporal.workflow.ActivityStub; import io.temporal.workflow.Functions; import io.temporal.workflow.Promise; import java.lang.reflect.Type; +import java.util.Objects; +import javax.annotation.Nullable; class LocalActivityStubImpl extends ActivityStubBase { protected final LocalActivityOptions options; @@ -31,14 +34,43 @@ private LocalActivityStubImpl( this.assertReadOnly = assertReadOnly; } + @Override + public R execute( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions invocationOptions, + Object... args) { + Objects.requireNonNull(invocationOptions, "invocationOptions"); + return getResult( + scheduleActivity( + activityName, resultClass, resultType, invocationOptions.getActivityId(), args), + resultClass); + } + @Override public Promise executeAsync( - String activityName, Class resultClass, Type resultType, Object... args) { + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions invocationOptions, + Object... args) { + Objects.requireNonNull(invocationOptions, "invocationOptions"); + return scheduleActivity( + activityName, resultClass, resultType, invocationOptions.getActivityId(), args); + } + + private Promise scheduleActivity( + String activityName, + Class resultClass, + Type resultType, + @Nullable String activityId, + Object... args) { this.assertReadOnly.apply(); return activityExecutor .executeLocalActivity( new WorkflowOutboundCallsInterceptor.LocalActivityInput<>( - activityName, resultClass, resultType, args, options, Header.empty())) + activityName, activityId, resultClass, resultType, args, options, Header.empty())) .getResult(); } } 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 065ce71428..4da6fda951 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 @@ -282,7 +282,12 @@ public ActivityOutput executeActivity(ActivityInput input) { Optional args = dataConverterWithActivityContext.toPayloads(input.getArgs()); ActivityOutput> output = - executeActivityOnce(input.getActivityName(), input.getOptions(), input.getHeader(), args); + executeActivityOnce( + input.getActivityName(), + input.getActivityId(), + input.getOptions(), + input.getHeader(), + args); // Avoid passing the input to the output handle as it causes the input to be retained for the // duration of the operation. @@ -307,9 +312,13 @@ public ActivityOutput executeActivity(ActivityInput input) { } private ActivityOutput> executeActivityOnce( - String activityTypeName, ActivityOptions options, Header header, Optional input) { + String activityTypeName, + @Nullable String activityId, + ActivityOptions options, + Header header, + Optional input) { ExecuteActivityParameters params = - constructExecuteActivityParameters(activityTypeName, options, header, input); + constructExecuteActivityParameters(activityTypeName, activityId, options, header, input); ActivityCallback callback = new ActivityCallback(); ReplayWorkflowContext.ScheduleActivityTaskOutput activityOutput = replayContext.scheduleActivityTask(params, callback::invoke); @@ -447,6 +456,7 @@ public LocalActivityOutput executeLocalActivity(LocalActivityInput inp WorkflowInternal.newCompletablePromise(); executeLocalActivityOverLocalRetryThreshold( input.getActivityName(), + input.getActivityId(), input.getOptions(), input.getHeader(), payloads, @@ -477,6 +487,7 @@ public LocalActivityOutput executeLocalActivity(LocalActivityInput inp public void executeLocalActivityOverLocalRetryThreshold( String activityTypeName, + @Nullable String activityId, LocalActivityOptions options, Header header, Optional input, @@ -487,6 +498,7 @@ public void executeLocalActivityOverLocalRetryThreshold( CompletablePromise> localExecutionResult = executeLocalActivityLocally( activityTypeName, + activityId, options, header, input, @@ -509,6 +521,7 @@ public void executeLocalActivityOverLocalRetryThreshold( unused -> { executeLocalActivityOverLocalRetryThreshold( activityTypeName, + activityId, options, header, input, @@ -539,6 +552,7 @@ public void executeLocalActivityOverLocalRetryThreshold( private CompletablePromise> executeLocalActivityLocally( String activityTypeName, + @Nullable String activityId, LocalActivityOptions options, Header header, Optional input, @@ -550,6 +564,7 @@ private CompletablePromise> executeLocalActivityLocally( ExecuteLocalActivityParameters params = constructExecuteLocalActivityParameters( activityTypeName, + activityId, options, header, input, @@ -569,7 +584,11 @@ private CompletablePromise> executeLocalActivityLocally( @SuppressWarnings("deprecation") private ExecuteActivityParameters constructExecuteActivityParameters( - String name, ActivityOptions options, Header header, Optional input) { + String name, + @Nullable String activityId, + ActivityOptions options, + Header header, + Optional input) { String taskQueue = options.getTaskQueue(); if (taskQueue == null) { taskQueue = replayContext.getTaskQueue(); @@ -589,6 +608,10 @@ private ExecuteActivityParameters constructExecuteActivityParameters( !options.isEagerExecutionDisabled() && Objects.equals(taskQueue, replayContext.getTaskQueue())); + if (activityId != null) { + attributes.setActivityId(activityId); + } + input.ifPresent(attributes::setInput); RetryOptions retryOptions = options.getRetryOptions(); if (retryOptions != null) { @@ -626,6 +649,7 @@ private ExecuteActivityParameters constructExecuteActivityParameters( private ExecuteLocalActivityParameters constructExecuteLocalActivityParameters( String name, + @Nullable String activityId, LocalActivityOptions options, Header header, Optional input, @@ -636,7 +660,8 @@ private ExecuteLocalActivityParameters constructExecuteLocalActivityParameters( PollActivityTaskQueueResponse.Builder activityTask = PollActivityTaskQueueResponse.newBuilder() - .setActivityId(this.replayContext.randomUUID().toString()) + .setActivityId( + activityId != null ? activityId : this.replayContext.randomUUID().toString()) .setWorkflowNamespace(this.replayContext.getNamespace()) .setWorkflowType(this.replayContext.getWorkflowType()) .setWorkflowExecution(this.replayContext.getWorkflowExecution()) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java index 84b1e91fd3..0a1c982f07 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java @@ -511,7 +511,7 @@ public static R executeActivity( getWorkflowOutboundInterceptor() .executeActivity( new WorkflowOutboundCallsInterceptor.ActivityInput<>( - name, resultClass, resultType, args, options, Header.empty())) + name, null, resultClass, resultType, args, options, Header.empty())) .getResult(); if (AsyncInternal.isAsync()) { AsyncInternal.setAsyncResult(result); @@ -520,6 +520,23 @@ public static R executeActivity( return result.get(); } + public static Promise executeActivityAsync( + ActivityInvocationOptions options, Functions.Proc invocation) { + assertNotReadOnly("schedule activity"); + return ActivityInvocationInternal.invokeAsync(options, invocation); + } + + public static R executeActivity( + ActivityInvocationOptions options, Functions.Func invocation) { + assertNotReadOnly("schedule activity"); + return ActivityInvocationInternal.invoke(options, invocation); + } + + public static void executeActivity(ActivityInvocationOptions options, Functions.Proc invocation) { + assertNotReadOnly("schedule activity"); + ActivityInvocationInternal.invoke(options, invocation); + } + public static void await(String reason, Supplier unblockCondition) throws DestroyWorkflowThreadError { assertNotReadOnly(reason); diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/ActivityInvocationOptions.java b/temporal-sdk/src/main/java/io/temporal/workflow/ActivityInvocationOptions.java new file mode 100644 index 0000000000..d3c1696ec0 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/workflow/ActivityInvocationOptions.java @@ -0,0 +1,83 @@ +package io.temporal.workflow; + +import io.temporal.common.Experimental; +import java.util.Objects; +import javax.annotation.Nullable; + +/** Options that apply to a single Workflow Activity or Local Activity invocation. */ +@Experimental +public final class ActivityInvocationOptions { + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(ActivityInvocationOptions options) { + return new Builder(options); + } + + public static final class Builder { + private String activityId; + + private Builder() {} + + private Builder(ActivityInvocationOptions options) { + if (options != null) { + this.activityId = options.activityId; + } + } + + /** + * Sets the identifier for this Activity or Local Activity invocation. + * + *

The identifier must be unique among open Activity Executions within the current Workflow + * Run. If it is not set, the SDK generates an identifier. + */ + public Builder setActivityId(String activityId) { + Objects.requireNonNull(activityId, "activityId"); + if (activityId.isEmpty()) { + throw new IllegalArgumentException("activityId must not be empty"); + } + this.activityId = activityId; + return this; + } + + public ActivityInvocationOptions build() { + return new ActivityInvocationOptions(activityId); + } + } + + private final String activityId; + + private ActivityInvocationOptions(String activityId) { + this.activityId = activityId; + } + + /** Returns the caller-supplied Activity ID, or {@code null} if the SDK should generate one. */ + @Nullable + public String getActivityId() { + return activityId; + } + + public Builder toBuilder() { + return new Builder(this); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + ActivityInvocationOptions that = (ActivityInvocationOptions) o; + return Objects.equals(activityId, that.activityId); + } + + @Override + public int hashCode() { + return Objects.hash(activityId); + } + + @Override + public String toString() { + return "ActivityInvocationOptions{" + "activityId='" + activityId + '\'' + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/ActivityStub.java b/temporal-sdk/src/main/java/io/temporal/workflow/ActivityStub.java index 0e5f8b3409..4a9f1eb511 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/ActivityStub.java +++ b/temporal-sdk/src/main/java/io/temporal/workflow/ActivityStub.java @@ -61,4 +61,68 @@ public interface ActivityStub { */ Promise executeAsync( String activityName, Class resultClass, Type resultType, Object... args); + + /** + * Executes an Activity with options that apply only to this invocation. Blocks until completion. + * + * @param activityName name of an Activity type to execute. + * @param resultClass expected return type of the Activity. + * @param options options for this invocation. + * @param args arguments of the Activity. + * @param return type. + * @return Activity result. + */ + R execute( + String activityName, Class resultClass, ActivityInvocationOptions options, Object... args); + + /** + * Executes an Activity with options that apply only to this invocation. Blocks until completion. + * + * @param activityName name of an Activity type to execute. + * @param resultClass expected return class of the Activity. + * @param resultType expected return type of the Activity. Differs from {@code resultClass} for + * generic types. + * @param options options for this invocation. + * @param args arguments of the Activity. + * @param return type. + * @return Activity result. + */ + R execute( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions options, + Object... args); + + /** + * Executes an Activity asynchronously with options that apply only to this invocation. + * + * @param activityName name of an Activity type to execute. + * @param resultClass expected return type of the Activity. + * @param options options for this invocation. + * @param args arguments of the Activity. + * @param return type. + * @return Promise to the Activity result. + */ + Promise executeAsync( + String activityName, Class resultClass, ActivityInvocationOptions options, Object... args); + + /** + * Executes an Activity asynchronously with options that apply only to this invocation. + * + * @param activityName name of an Activity type to execute. + * @param resultClass expected return class of the Activity. + * @param resultType expected return type of the Activity. Differs from {@code resultClass} for + * generic types. + * @param options options for this invocation. + * @param args arguments of the Activity. + * @param return type. + * @return Promise to the Activity result. + */ + Promise executeAsync( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions options, + Object... args); } diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java index d04a617d5d..40ecad495c 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java +++ b/temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java @@ -100,6 +100,298 @@ public static ActivityStub newUntypedActivityStub(ActivityOptions options) { return WorkflowInternal.newUntypedActivityStub(options); } + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func activity, ActivityInvocationOptions options) { + return WorkflowInternal.executeActivity(options, activity::apply); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func1 activity, ActivityInvocationOptions options, A1 arg1) { + return WorkflowInternal.executeActivity(options, () -> activity.apply(arg1)); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func2 activity, ActivityInvocationOptions options, A1 arg1, A2 arg2) { + return WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2)); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func3 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3) { + return WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2, arg3)); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func4 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4) { + return WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2, arg3, arg4)); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func5 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5) { + return WorkflowInternal.executeActivity( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5)); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func6 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6) { + return WorkflowInternal.executeActivity( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5, arg6)); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity(Functions.Proc activity, ActivityInvocationOptions options) { + WorkflowInternal.executeActivity(options, activity); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc1 activity, ActivityInvocationOptions options, A1 arg1) { + WorkflowInternal.executeActivity(options, () -> activity.apply(arg1)); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc2 activity, ActivityInvocationOptions options, A1 arg1, A2 arg2) { + WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2)); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc3 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3) { + WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2, arg3)); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc4 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4) { + WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2, arg3, arg4)); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc5 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5) { + WorkflowInternal.executeActivity(options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5)); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc6 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6) { + WorkflowInternal.executeActivity( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5, arg6)); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func activity, ActivityInvocationOptions options) { + return WorkflowInternal.executeActivityAsync(options, activity::apply); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func1 activity, ActivityInvocationOptions options, A1 arg1) { + return WorkflowInternal.executeActivityAsync(options, () -> activity.apply(arg1)); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func2 activity, ActivityInvocationOptions options, A1 arg1, A2 arg2) { + return WorkflowInternal.executeActivityAsync(options, () -> activity.apply(arg1, arg2)); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func3 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3) { + return WorkflowInternal.executeActivityAsync(options, () -> activity.apply(arg1, arg2, arg3)); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func4 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4) { + return WorkflowInternal.executeActivityAsync( + options, () -> activity.apply(arg1, arg2, arg3, arg4)); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func5 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5) { + return WorkflowInternal.executeActivityAsync( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5)); + } + + /** Starts an Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Func6 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6) { + return WorkflowInternal.executeActivityAsync( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5, arg6)); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc activity, ActivityInvocationOptions options) { + return WorkflowInternal.executeActivityAsync(options, activity); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc1 activity, ActivityInvocationOptions options, A1 arg1) { + return WorkflowInternal.executeActivityAsync(options, () -> activity.apply(arg1)); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc2 activity, ActivityInvocationOptions options, A1 arg1, A2 arg2) { + return WorkflowInternal.executeActivityAsync(options, () -> activity.apply(arg1, arg2)); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc3 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3) { + return WorkflowInternal.executeActivityAsync(options, () -> activity.apply(arg1, arg2, arg3)); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc4 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4) { + return WorkflowInternal.executeActivityAsync( + options, () -> activity.apply(arg1, arg2, arg3, arg4)); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc5 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5) { + return WorkflowInternal.executeActivityAsync( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5)); + } + + /** Starts a void Activity with options that apply only to this invocation. */ + @Experimental + public static Promise executeActivityAsync( + Functions.Proc6 activity, + ActivityInvocationOptions options, + A1 arg1, + A2 arg2, + A3 arg3, + A4 arg4, + A5 arg5, + A6 arg6) { + return WorkflowInternal.executeActivityAsync( + options, () -> activity.apply(arg1, arg2, arg3, arg4, arg5, arg6)); + } + /** * Creates client stub to local activities that implement given interface. * diff --git a/temporal-sdk/src/main/java/io/temporal/workflow/package-info.java b/temporal-sdk/src/main/java/io/temporal/workflow/package-info.java index f8372c2e58..9d393ff7b8 100644 --- a/temporal-sdk/src/main/java/io/temporal/workflow/package-info.java +++ b/temporal-sdk/src/main/java/io/temporal/workflow/package-info.java @@ -110,6 +110,22 @@ * } * * + * Options that identify a single Activity or Local Activity invocation are not stored on the + * reusable stub. Use {@link io.temporal.workflow.ActivityInvocationOptions} with {@link + * io.temporal.workflow.Workflow#executeActivity(Functions.Func1, ActivityInvocationOptions, + * Object)} or its asynchronous variant to supply an optional Activity ID for one invocation. + * + * 


+ * ActivityInvocationOptions invocationOptions = ActivityInvocationOptions.newBuilder()
+ *     .setActivityId("charge-" + order.getId())
+ *     .build();
+ *
+ * Receipt receipt = Workflow.executeActivity(
+ *     activities::charge,
+ *     invocationOptions,
+ *     order);
+ *
+ * *

Calling Activities Asynchronously

* * Sometimes workflows need to perform certain operations in parallel. The {@link diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/ActivityInvocationOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/ActivityInvocationOptionsTest.java new file mode 100644 index 0000000000..956f47a992 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/ActivityInvocationOptionsTest.java @@ -0,0 +1,37 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +import org.junit.Test; + +public class ActivityInvocationOptionsTest { + + @Test + public void setActivityIdRejectsNull() { + NullPointerException e = + assertThrows( + NullPointerException.class, + () -> ActivityInvocationOptions.newBuilder().setActivityId(null)); + assertEquals("activityId", e.getMessage()); + } + + @Test + public void setActivityIdRejectsEmpty() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> ActivityInvocationOptions.newBuilder().setActivityId("")); + assertEquals("activityId must not be empty", e.getMessage()); + } + + @Test + public void newBuilderCopiesActivityId() { + ActivityInvocationOptions original = + ActivityInvocationOptions.newBuilder().setActivityId("activity-123").build(); + + ActivityInvocationOptions copy = ActivityInvocationOptions.newBuilder(original).build(); + + assertEquals("activity-123", copy.getActivityId()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityInvocationIdTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityInvocationIdTest.java new file mode 100644 index 0000000000..3715e6402a --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityInvocationIdTest.java @@ -0,0 +1,431 @@ +package io.temporal.workflow.activityTests; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import com.google.common.reflect.TypeToken; +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInfo; +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.WorkflowExecution; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.client.WorkflowStub; +import io.temporal.common.RetryOptions; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.testing.WorkflowReplayer; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.ActivityInvocationOptions; +import io.temporal.workflow.ActivityStub; +import io.temporal.workflow.Promise; +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.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +public class ActivityInvocationIdTest { + private static final Duration ACTIVITY_TIMEOUT = Duration.ofSeconds(5); + private static final ActivityOptions ACTIVITY_OPTIONS = + ActivityOptions.newBuilder().setStartToCloseTimeout(ACTIVITY_TIMEOUT).build(); + private static final LocalActivityOptions LOCAL_ACTIVITY_OPTIONS = + LocalActivityOptions.newBuilder().setStartToCloseTimeout(ACTIVITY_TIMEOUT).build(); + private static final InvocationIdActivitiesImpl ACTIVITIES = new InvocationIdActivitiesImpl(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TestWorkflowImpl.class, GenericWorkflowImpl.class) + .setActivityImplementations(ACTIVITIES) + .build(); + + @Before + public void setUp() { + TestWorkflowImpl.configuredActivityId = "replay-activity"; + } + + @Test + public void typedExecuteActivityRecordsSuppliedIdInActivityInfoAndHistory() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = workflow.execute(Invocation.TYPED_SYNC, "typed-sync-activity"); + + assertEquals(Collections.singletonList("typed-sync-activity"), result); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void typedExecuteActivityAsyncPreservesGenericReturnType() { + GenericWorkflow workflow = newWorkflow(GenericWorkflow.class); + List values = Arrays.asList(UUID.randomUUID(), UUID.randomUUID()); + + List result = + workflow.execute(GenericInvocation.TYPED_ASYNC, values, "typed-async-generic"); + + assertEquals(values, result); + assertEquals(Collections.singletonList("typed-async-generic"), scheduledActivityIds(workflow)); + } + + @Test + public void untypedExecuteOverloadSupportsGenericResultType() { + GenericWorkflow workflow = newWorkflow(GenericWorkflow.class); + List values = Arrays.asList(UUID.randomUUID(), UUID.randomUUID()); + + List result = workflow.execute(GenericInvocation.UNTYPED_SYNC, values, "untyped-generic"); + + assertEquals(values, result); + assertEquals(Collections.singletonList("untyped-generic"), scheduledActivityIds(workflow)); + } + + @Test + public void concurrentTypedInvocationsUsingSameStubKeepDistinctIds() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = + workflow.execute( + Invocation.CONCURRENT_TYPED, "concurrent-activity-a", "concurrent-activity-b"); + + assertEquals(Arrays.asList("concurrent-activity-a", "concurrent-activity-b"), result); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void concurrentUntypedInvocationsUsingSameStubKeepDistinctIds() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = + workflow.execute( + Invocation.CONCURRENT_UNTYPED, "untyped-concurrent-a", "untyped-concurrent-b"); + + assertEquals(Arrays.asList("untyped-concurrent-a", "untyped-concurrent-b"), result); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void typedVoidActivityUsesSuppliedId() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = workflow.execute(Invocation.TYPED_VOID, "typed-void-activity"); + + assertEquals(Collections.singletonList("typed-void-activity"), result); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void executeActivityWithoutSuppliedIdPreservesGeneratedFallback() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = workflow.execute(Invocation.OMITTED_ID); + + assertFalse(result.get(0).isEmpty()); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void completedActivityIdCanBeReused() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = workflow.execute(Invocation.REUSED_ID); + + assertEquals(Arrays.asList("reused-activity", "reused-activity"), result); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void executeActivitySupportsLocalActivityMethodReference() throws Exception { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List activityIds = workflow.execute(Invocation.LOCAL_REFERENCES); + + assertEquals( + Arrays.asList("typed-local-activity", "untyped-local-activity"), activityIds.subList(0, 2)); + assertFalse(activityIds.get(2).isEmpty()); + WorkflowReplayer.replayWorkflowExecution(history(workflow), TestWorkflowImpl.class); + } + + @Test + public void localActivityIdIsPreservedAcrossTimerBackedRetry() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + assertEquals( + Collections.singletonList("local-retry-activity"), + workflow.execute(Invocation.LOCAL_RETRY)); + } + + @Test + public void executeActivitySupportsSingleActivityLambda() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = workflow.execute(Invocation.LAMBDA); + + assertEquals(Collections.singletonList("lambda-activity"), result); + assertEquals(result, scheduledActivityIds(workflow)); + } + + @Test + public void executeActivityPreservesRemoteStubValidationFailure() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + assertTrue( + workflow + .execute(Invocation.MISSING_TIMEOUT) + .get(0) + .contains("Both StartToCloseTimeout and ScheduleToCloseTimeout aren't specified")); + } + + @Test + public void replaySucceedsWhenExplicitActivityIdIsUnchanged() throws Exception { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + assertEquals(Collections.singletonList("replay-activity"), workflow.execute(Invocation.REPLAY)); + + WorkflowReplayer.replayWorkflowExecution(history(workflow), TestWorkflowImpl.class); + } + + @Test + public void replayFailsWhenExplicitActivityIdChanges() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + assertEquals(Collections.singletonList("replay-activity"), workflow.execute(Invocation.REPLAY)); + WorkflowExecutionHistory history = history(workflow); + TestWorkflowImpl.configuredActivityId = "replay-activity-changed"; + + assertThrows( + RuntimeException.class, + () -> WorkflowReplayer.replayWorkflowExecution(history, TestWorkflowImpl.class)); + } + + private T newWorkflow(Class workflowInterface) { + return testWorkflowRule.newWorkflowStubTimeoutOptions(workflowInterface); + } + + private List scheduledActivityIds(Object workflow) { + return history(workflow).getEvents().stream() + .filter(HistoryEvent::hasActivityTaskScheduledEventAttributes) + .map(event -> event.getActivityTaskScheduledEventAttributes().getActivityId()) + .collect(Collectors.toList()); + } + + private WorkflowExecutionHistory history(Object workflow) { + WorkflowExecution execution = WorkflowStub.fromTyped(workflow).getExecution(); + return testWorkflowRule + .getWorkflowClient() + .fetchHistory(execution.getWorkflowId(), execution.getRunId()); + } + + public enum Invocation { + TYPED_SYNC, + CONCURRENT_TYPED, + CONCURRENT_UNTYPED, + TYPED_VOID, + OMITTED_ID, + REUSED_ID, + LOCAL_REFERENCES, + LOCAL_RETRY, + LAMBDA, + MISSING_TIMEOUT, + REPLAY + } + + @WorkflowInterface + public interface TestWorkflow { + @WorkflowMethod + List execute(Invocation invocation, String... activityIds); + } + + public static class TestWorkflowImpl implements TestWorkflow { + private static volatile String configuredActivityId = "replay-activity"; + + private final InvocationIdActivities activities = + Workflow.newActivityStub(InvocationIdActivities.class, ACTIVITY_OPTIONS); + private final ActivityStub untypedActivities = + Workflow.newUntypedActivityStub(ACTIVITY_OPTIONS); + private final InvocationIdActivities localActivities = + Workflow.newLocalActivityStub(InvocationIdActivities.class, LOCAL_ACTIVITY_OPTIONS); + private final ActivityStub untypedLocalActivities = + Workflow.newUntypedLocalActivityStub(LOCAL_ACTIVITY_OPTIONS); + private final InvocationIdActivities activitiesWithoutTimeout = + Workflow.newActivityStub( + InvocationIdActivities.class, ActivityOptions.newBuilder().build()); + private final InvocationIdActivities retryingLocalActivities = + Workflow.newLocalActivityStub( + InvocationIdActivities.class, + LocalActivityOptions.newBuilder() + .setStartToCloseTimeout(ACTIVITY_TIMEOUT) + .setLocalRetryThreshold(Duration.ofMillis(1)) + .setRetryOptions( + RetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(10)) + .setMaximumAttempts(2) + .build()) + .build()); + + @Override + public List execute(Invocation invocation, String... activityIds) { + switch (invocation) { + case TYPED_SYNC: + return Collections.singletonList( + Workflow.executeActivity( + activities::recordActivityId, invocationOptions(activityIds[0]))); + case CONCURRENT_TYPED: + { + Promise first = + Workflow.executeActivityAsync( + activities::recordActivityId, invocationOptions(activityIds[0])); + Promise second = + Workflow.executeActivityAsync( + activities::recordActivityId, invocationOptions(activityIds[1])); + return Arrays.asList(first.get(), second.get()); + } + case CONCURRENT_UNTYPED: + { + Promise first = + untypedActivities.executeAsync( + "RecordActivityId", String.class, invocationOptions(activityIds[0])); + Promise second = + untypedActivities.executeAsync( + "RecordActivityId", String.class, invocationOptions(activityIds[1])); + return Arrays.asList(first.get(), second.get()); + } + case TYPED_VOID: + Workflow.executeActivity(activities::recordVoid, invocationOptions(activityIds[0])); + return Collections.singletonList(activityIds[0]); + case OMITTED_ID: + return Collections.singletonList( + Workflow.executeActivity( + activities::recordActivityId, ActivityInvocationOptions.newBuilder().build())); + case REUSED_ID: + { + ActivityInvocationOptions options = invocationOptions("reused-activity"); + String first = Workflow.executeActivity(activities::recordActivityId, options); + String second = Workflow.executeActivity(activities::recordActivityId, options); + return Arrays.asList(first, second); + } + case LOCAL_REFERENCES: + { + String typed = + Workflow.executeActivity( + localActivities::recordActivityId, invocationOptions("typed-local-activity")); + String untyped = + untypedLocalActivities.execute( + "RecordActivityId", String.class, invocationOptions("untyped-local-activity")); + String generated = + Workflow.executeActivity( + localActivities::recordActivityId, + ActivityInvocationOptions.newBuilder().build()); + return Arrays.asList(typed, untyped, generated); + } + case LOCAL_RETRY: + return Collections.singletonList( + Workflow.executeActivity( + retryingLocalActivities::failOnceAndReturnActivityId, + invocationOptions("local-retry-activity"))); + case LAMBDA: + return Collections.singletonList( + Workflow.executeActivity( + () -> activities.recordActivityId(), invocationOptions("lambda-activity"))); + case MISSING_TIMEOUT: + try { + Workflow.executeActivity( + activitiesWithoutTimeout::recordActivityId, invocationOptions("missing-timeout")); + return Collections.singletonList("unexpected success"); + } catch (IllegalArgumentException e) { + return Collections.singletonList(e.getMessage()); + } + case REPLAY: + return Collections.singletonList( + Workflow.executeActivity( + activities::recordActivityId, invocationOptions(configuredActivityId))); + } + throw new IllegalArgumentException("Unknown invocation: " + invocation); + } + + private static ActivityInvocationOptions invocationOptions(String activityId) { + return ActivityInvocationOptions.newBuilder().setActivityId(activityId).build(); + } + } + + public enum GenericInvocation { + TYPED_ASYNC, + UNTYPED_SYNC + } + + @WorkflowInterface + public interface GenericWorkflow { + @WorkflowMethod + List execute(GenericInvocation invocation, List values, String activityId); + } + + public static class GenericWorkflowImpl implements GenericWorkflow { + private static final Type UUID_LIST_TYPE = new TypeToken>() {}.getType(); + + private final InvocationIdActivities activities = + Workflow.newActivityStub(InvocationIdActivities.class, ACTIVITY_OPTIONS); + private final ActivityStub untypedActivities = + Workflow.newUntypedActivityStub(ACTIVITY_OPTIONS); + + @Override + public List execute(GenericInvocation invocation, List values, String activityId) { + ActivityInvocationOptions options = TestWorkflowImpl.invocationOptions(activityId); + switch (invocation) { + case TYPED_ASYNC: + return Workflow.executeActivityAsync(activities::echoUuidList, options, values).get(); + case UNTYPED_SYNC: + return untypedActivities.execute( + "EchoUuidList", List.class, UUID_LIST_TYPE, options, values); + } + throw new IllegalArgumentException("Unknown invocation: " + invocation); + } + } + + @ActivityInterface + public interface InvocationIdActivities { + @ActivityMethod(name = "RecordActivityId") + String recordActivityId(); + + @ActivityMethod(name = "FailOnceAndReturnActivityId") + String failOnceAndReturnActivityId(); + + @ActivityMethod(name = "EchoUuidList") + List echoUuidList(List values); + + @ActivityMethod(name = "RecordVoid") + void recordVoid(); + } + + public static class InvocationIdActivitiesImpl implements InvocationIdActivities { + @Override + public String recordActivityId() { + return Activity.getExecutionContext().getInfo().getActivityId(); + } + + @Override + public String failOnceAndReturnActivityId() { + ActivityInfo info = Activity.getExecutionContext().getInfo(); + if (info.getAttempt() == 1) { + throw new RuntimeException("intentional first-attempt failure"); + } + return info.getActivityId(); + } + + @Override + public List echoUuidList(List values) { + return values; + } + + @Override + public void recordVoid() {} + } +}