From 01d272538ad7c87d3ba928fabbe220d92241e192 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Wed, 26 Aug 2026 14:33:00 -0700 Subject: [PATCH] Add support for setting ActivityID --- .../WorkflowOutboundCallsInterceptor.java | 38 ++ .../sync/ActivityInvocationHandler.java | 36 +- .../sync/ActivityInvocationInternal.java | 77 +++ .../internal/sync/ActivityStubBase.java | 47 ++ .../internal/sync/ActivityStubImpl.java | 40 +- .../sync/LocalActivityInvocationHandler.java | 24 +- .../internal/sync/LocalActivityStubImpl.java | 29 +- .../internal/sync/SyncWorkflowContext.java | 35 +- .../internal/sync/WorkflowInternal.java | 6 + .../workflow/ActivityInvocationOptions.java | 120 +++++ .../io/temporal/workflow/ActivityStub.java | 64 +++ .../java/io/temporal/workflow/Workflow.java | 289 ++++++++++ .../io/temporal/workflow/package-info.java | 16 + .../ActivityInvocationOptionsTest.java | 51 ++ .../ActivityInvocationIdTest.java | 493 ++++++++++++++++++ 15 files changed, 1348 insertions(+), 17 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationInternal.java create mode 100644 temporal-sdk/src/main/java/io/temporal/workflow/ActivityInvocationOptions.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/ActivityInvocationOptionsTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityInvocationIdTest.java 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..d8ea727b2e 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,6 +41,7 @@ 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; @@ -54,7 +55,19 @@ public ActivityInput( 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 +79,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,6 +126,7 @@ 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; @@ -120,7 +140,19 @@ public LocalActivityInput( 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 +164,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..287b97a9f8 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,8 +4,10 @@ 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 io.temporal.workflow.Promise; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.util.HashMap; @@ -46,22 +48,44 @@ private ActivityInvocationHandler( @Override protected Function getActivityFunc( Method method, MethodRetry methodRetry, String activityName) { - Function function; ActivityOptions merged = ActivityOptions.newBuilder(options) .mergeActivityOptions(this.activityMethodOptions.get(activityName)) .mergeMethodRetry(methodRetry) .build(); - if (merged.getStartToCloseTimeout() == null && merged.getScheduleToCloseTimeout() == null) { + + if (ActivityInvocationInternal.isActive()) { + ActivityInvocationOptions invocationOptions = ActivityInvocationInternal.consumeOptions(); + ActivityStub stub = + newStub(ActivityStubImpl.resolveOptions(merged, invocationOptions), merged, activityName); + return (a) -> { + Promise result = + stub.executeAsync( + activityName, + method.getReturnType(), + method.getGenericReturnType(), + invocationOptions, + a); + ActivityInvocationInternal.setResult(result); + return null; + }; + } + + ActivityStub stub = newStub(merged, merged, activityName); + return (a) -> + stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a); + } + + private ActivityStub newStub( + ActivityOptions effectiveOptions, ActivityOptions stubOptions, String activityName) { + if (effectiveOptions.getStartToCloseTimeout() == null + && effectiveOptions.getScheduleToCloseTimeout() == null) { throw new IllegalArgumentException( "Both StartToCloseTimeout and ScheduleToCloseTimeout aren't specified for " + activityName + " activity. Please set at least one of the above through the ActivityStub or WorkflowImplementationOptions."); } - ActivityStub stub = ActivityStubImpl.newInstance(merged, activityExecutor, assertReadOnly); - function = - (a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a); - return function; + return ActivityStubImpl.newInstance(stubOptions, activityExecutor, assertReadOnly); } @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..0a790c45e3 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityInvocationInternal.java @@ -0,0 +1,77 @@ +package io.temporal.internal.sync; + +import io.temporal.workflow.ActivityInvocationOptions; +import io.temporal.workflow.Functions; +import io.temporal.workflow.Promise; +import java.util.Objects; + +/** Captures one typed Activity proxy invocation and its result Promise. */ +final class ActivityInvocationInternal { + + private static final ThreadLocal invocation = new ThreadLocal<>(); + + private ActivityInvocationInternal() {} + + static Promise invoke( + ActivityInvocationOptions options, Functions.Proc invocationFunction) { + if (invocation.get() != null) { + throw new IllegalStateException("Already invoking an Activity with invocation options"); + } + + State state = new State(Objects.requireNonNull(options, "options")); + invocation.set(state); + try { + invocationFunction.apply(); + return state.getResult(); + } finally { + invocation.remove(); + } + } + + static ActivityInvocationOptions consumeOptions() { + State state = invocation.get(); + if (state == null) { + throw new IllegalStateException("Not invoking an Activity with invocation options"); + } + if (state.consumed) { + throw new IllegalStateException("ActivityInvocationOptions can apply to only one invocation"); + } + state.consumed = true; + return state.options; + } + + static boolean isActive() { + return invocation.get() != null; + } + + static void setResult(Promise result) { + State state = invocation.get(); + if (state == null) { + throw new IllegalStateException("Not invoking an Activity with invocation options"); + } + if (state.result != null) { + throw new IllegalStateException("ActivityInvocationOptions can apply to only one invocation"); + } + state.result = Objects.requireNonNull(result, "result"); + } + + private static final class State { + private final ActivityInvocationOptions options; + private boolean consumed; + private Promise result; + + private State(ActivityInvocationOptions options) { + this.options = options; + } + + @SuppressWarnings("unchecked") + private Promise getResult() { + if (!consumed || result == null) { + throw new IllegalArgumentException( + "activityMethod must invoke an Activity stub created through Workflow.newActivityStub " + + "or Workflow.newLocalActivityStub"); + } + return (Promise) result; + } + } +} 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..82e00dcb8b 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,6 +2,7 @@ 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; @@ -40,4 +41,50 @@ public Promise executeAsync(String activityName, Class resultClass, Ob @Override public abstract Promise executeAsync( String activityName, Class resultClass, Type resultType, Object... args); + + @Override + public R execute( + String activityName, + Class resultClass, + ActivityInvocationOptions options, + Object... args) { + return execute(activityName, resultClass, resultClass, options, args); + } + + @Override + public R execute( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions options, + Object... args) { + Promise result = executeAsync(activityName, resultClass, resultType, options, args); + if (AsyncInternal.isAsync()) { + AsyncInternal.setAsyncResult(result); + return Defaults.defaultValue(resultClass); + } + try { + return result.get(); + } catch (ActivityFailure e) { + e.setStackTrace(Thread.currentThread().getStackTrace()); + throw e; + } + } + + @Override + 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, + 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..f86f46dc05 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,10 +3,12 @@ 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; @@ -31,14 +33,50 @@ static ActivityStub newInstance( this.assertReadOnly = assertReadOnly; } + static ActivityOptions resolveOptions( + ActivityOptions options, ActivityInvocationOptions invocationOptions) { + ActivityOptions invocationActivityOptions = invocationOptions.getActivityOptions(); + if (invocationActivityOptions == null) { + return options; + } + return ActivityOptions.newBuilder(invocationActivityOptions).validateAndBuildWithDefaults(); + } + @Override public Promise executeAsync( String activityName, Class resultClass, Type resultType, Object... args) { + return executeAsyncInternal(activityName, resultClass, resultType, null, options, args); + } + + @Override + public Promise executeAsync( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions invocationOptions, + Object... args) { + Objects.requireNonNull(invocationOptions, "invocationOptions"); + return executeAsyncInternal( + activityName, + resultClass, + resultType, + invocationOptions.getActivityId(), + resolveOptions(options, invocationOptions), + args); + } + + private Promise executeAsyncInternal( + String activityName, + Class resultClass, + Type resultType, + String activityId, + ActivityOptions options, + Object... args) { this.assertReadOnly.apply(); return activityExecutor .executeActivity( new WorkflowOutboundCallsInterceptor.ActivityInput<>( - 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/LocalActivityInvocationHandler.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityInvocationHandler.java index 5b173d33f3..aea416e3e9 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,8 +4,10 @@ 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 io.temporal.workflow.Promise; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.util.HashMap; @@ -47,7 +49,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 +56,24 @@ public Function getActivityFunc( .build(); ActivityStub stub = LocalActivityStubImpl.newInstance(mergedOptions, activityExecutor, assertReadOnly); - function = - (a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a); - return function; + + if (ActivityInvocationInternal.isActive()) { + ActivityInvocationOptions invocationOptions = ActivityInvocationInternal.consumeOptions(); + return (a) -> { + Promise result = + stub.executeAsync( + activityName, + method.getReturnType(), + method.getGenericReturnType(), + invocationOptions, + a); + ActivityInvocationInternal.setResult(result); + return null; + }; + } + + return (a) -> + stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), 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..aaa57fedde 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; @@ -34,11 +37,35 @@ private LocalActivityStubImpl( @Override public Promise executeAsync( String activityName, Class resultClass, Type resultType, Object... args) { + return executeAsyncInternal(activityName, resultClass, resultType, null, args); + } + + @Override + public Promise executeAsync( + String activityName, + Class resultClass, + Type resultType, + ActivityInvocationOptions invocationOptions, + Object... args) { + Objects.requireNonNull(invocationOptions, "invocationOptions"); + if (invocationOptions.getActivityOptions() != null) { + throw new IllegalArgumentException("ActivityOptions are not supported for Local Activities"); + } + return executeAsyncInternal( + activityName, resultClass, resultType, invocationOptions.getActivityId(), args); + } + + private Promise executeAsyncInternal( + 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..0a32326a0e 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 @@ -520,6 +520,12 @@ public static R executeActivity( return result.get(); } + public static Promise executeActivityAsync( + ActivityInvocationOptions options, Functions.Proc invocation) { + assertNotReadOnly("schedule activity"); + return 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..ee49e73ca8 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/workflow/ActivityInvocationOptions.java @@ -0,0 +1,120 @@ +package io.temporal.workflow; + +import io.temporal.activity.ActivityOptions; +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); + } + + /** Creates a builder with non-local Activity options that apply only to this invocation. */ + public static Builder newBuilder(ActivityOptions activityOptions) { + return new Builder().setActivityOptions(activityOptions); + } + + public static final class Builder { + private String activityId; + private ActivityOptions activityOptions; + + private Builder() {} + + private Builder(ActivityInvocationOptions options) { + if (options != null) { + this.activityId = options.activityId; + this.activityOptions = options.activityOptions; + } + } + + /** + * 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; + } + + /** + * Sets Activity options to use instead of the reusable stub options for this invocation. + * + *

These options completely replace stub and method-specific options for this invocation. + * They cannot be used with Local Activities. + */ + public Builder setActivityOptions(ActivityOptions activityOptions) { + this.activityOptions = Objects.requireNonNull(activityOptions, "activityOptions"); + return this; + } + + public ActivityInvocationOptions build() { + return new ActivityInvocationOptions(activityId, activityOptions); + } + } + + private final String activityId; + private final ActivityOptions activityOptions; + + private ActivityInvocationOptions(String activityId, ActivityOptions activityOptions) { + this.activityId = activityId; + this.activityOptions = activityOptions; + } + + /** Returns the caller-supplied Activity ID, or {@code null} if the SDK should generate one. */ + @Nullable + public String getActivityId() { + return activityId; + } + + /** + * Returns replacement options for this non-local Activity invocation, or {@code null} if none + * were supplied. + */ + @Nullable + public ActivityOptions getActivityOptions() { + return activityOptions; + } + + 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) + && Objects.equals(activityOptions, that.activityOptions); + } + + @Override + public int hashCode() { + return Objects.hash(activityId, activityOptions); + } + + @Override + public String toString() { + return "ActivityInvocationOptions{" + + "activityId='" + + activityId + + '\'' + + ", activityOptions=" + + activityOptions + + '}'; + } +} 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..cda8b2a451 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,295 @@ 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 executeActivityAsync(activity, options).get(); + } + + /** Executes an Activity with options that apply only to this invocation. */ + @Experimental + public static R executeActivity( + Functions.Func1 activity, ActivityInvocationOptions options, A1 arg1) { + return executeActivityAsync(activity, options, arg1).get(); + } + + /** 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 executeActivityAsync(activity, options, arg1, arg2).get(); + } + + /** 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 executeActivityAsync(activity, options, arg1, arg2, arg3).get(); + } + + /** 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 executeActivityAsync(activity, options, arg1, arg2, arg3, arg4).get(); + } + + /** 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 executeActivityAsync(activity, options, arg1, arg2, arg3, arg4, arg5).get(); + } + + /** 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 executeActivityAsync(activity, options, arg1, arg2, arg3, arg4, arg5, arg6).get(); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity(Functions.Proc activity, ActivityInvocationOptions options) { + executeActivityAsync(activity, options).get(); + } + + /** Executes a void Activity with options that apply only to this invocation. */ + @Experimental + public static void executeActivity( + Functions.Proc1 activity, ActivityInvocationOptions options, A1 arg1) { + executeActivityAsync(activity, options, arg1).get(); + } + + /** 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) { + executeActivityAsync(activity, options, arg1, arg2).get(); + } + + /** 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) { + executeActivityAsync(activity, options, arg1, arg2, arg3).get(); + } + + /** 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) { + executeActivityAsync(activity, options, arg1, arg2, arg3, arg4).get(); + } + + /** 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) { + executeActivityAsync(activity, options, arg1, arg2, arg3, arg4, arg5).get(); + } + + /** 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) { + executeActivityAsync(activity, options, arg1, arg2, arg3, arg4, arg5, arg6).get(); + } + + /** 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..c3b12d7b53 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/ActivityInvocationOptionsTest.java @@ -0,0 +1,51 @@ +package io.temporal.workflow; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +import io.temporal.activity.ActivityOptions; +import java.time.Duration; +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() { + ActivityOptions activityOptions = + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(1)).build(); + ActivityInvocationOptions original = + ActivityInvocationOptions.newBuilder(activityOptions).setActivityId("activity-123").build(); + + ActivityInvocationOptions copy = ActivityInvocationOptions.newBuilder(original).build(); + + assertEquals("activity-123", copy.getActivityId()); + assertEquals(activityOptions, copy.getActivityOptions()); + } + + @Test + public void setActivityOptionsRejectsNull() { + NullPointerException e = + assertThrows( + NullPointerException.class, + () -> ActivityInvocationOptions.newBuilder().setActivityOptions(null)); + assertEquals("activityOptions", e.getMessage()); + } +} 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..830707780c --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/ActivityInvocationIdTest.java @@ -0,0 +1,493 @@ +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 localActivityRejectsRemoteActivityOptions() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + assertEquals( + Collections.singletonList("ActivityOptions are not supported for Local Activities"), + workflow.execute(Invocation.INVALID_LOCAL_OPTIONS)); + } + + @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 invocationActivityOptionsOverrideTypedAndUntypedStubOptions() { + TestWorkflow workflow = newWorkflow(TestWorkflow.class); + + List result = + workflow.execute(Invocation.OPTIONS_OVERRIDE, "typed-options", "untyped-options"); + + assertEquals(Arrays.asList("typed-options", "untyped-options"), result); + assertEquals(result, scheduledActivityIds(workflow)); + history(workflow).getEvents().stream() + .filter(HistoryEvent::hasActivityTaskScheduledEventAttributes) + .forEach( + event -> + assertEquals( + ACTIVITY_TIMEOUT.getSeconds(), + event + .getActivityTaskScheduledEventAttributes() + .getStartToCloseTimeout() + .getSeconds())); + } + + @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, + INVALID_LOCAL_OPTIONS, + LAMBDA, + MISSING_TIMEOUT, + OPTIONS_OVERRIDE, + 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 ActivityStub untypedActivitiesWithoutTimeout = + Workflow.newUntypedActivityStub(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 INVALID_LOCAL_OPTIONS: + try { + Workflow.executeActivity( + localActivities::recordActivityId, + ActivityInvocationOptions.newBuilder(ACTIVITY_OPTIONS) + .setActivityId("invalid-local-activity") + .build()); + return Collections.singletonList("unexpected success"); + } catch (IllegalArgumentException e) { + return Collections.singletonList(e.getMessage()); + } + 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 OPTIONS_OVERRIDE: + { + String typed = + Workflow.executeActivity( + activitiesWithoutTimeout::recordActivityId, + ActivityInvocationOptions.newBuilder(ACTIVITY_OPTIONS) + .setActivityId(activityIds[0]) + .build()); + String untyped = + untypedActivitiesWithoutTimeout.execute( + "RecordActivityId", + String.class, + ActivityInvocationOptions.newBuilder(ACTIVITY_OPTIONS) + .setActivityId(activityIds[1]) + .build()); + return Arrays.asList(typed, untyped); + } + 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() {} + } +}