diff --git a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/StandaloneActivityClientTracingTest.java b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/StandaloneActivityClientTracingTest.java index c852175196..74919024d5 100644 --- a/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/StandaloneActivityClientTracingTest.java +++ b/contrib/temporal-opentracing/src/test/java/io/temporal/opentracing/StandaloneActivityClientTracingTest.java @@ -7,6 +7,7 @@ import io.opentracing.util.ThreadLocalScopeManager; import io.temporal.api.workflowservice.v1.CountActivityExecutionsResponse; import io.temporal.client.ActivityExecutionCount; +import io.temporal.client.DescribeActivityOptions; import io.temporal.client.StartActivityOptions; import io.temporal.common.interceptors.ActivityClientCallsInterceptor; import io.temporal.common.interceptors.ActivityClientCallsInterceptorBase; @@ -108,7 +109,8 @@ public void testManagementCallsDoNotCreateSpans() throws TimeoutException { new ActivityClientCallsInterceptor.GetActivityResultInput<>( "act-result-async", null, String.class)); interceptor.describeActivity( - new ActivityClientCallsInterceptor.DescribeActivityInput("act-desc", null)); + new ActivityClientCallsInterceptor.DescribeActivityInput( + "act-desc", null, DescribeActivityOptions.getDefaultInstance())); interceptor.cancelActivity( new ActivityClientCallsInterceptor.CancelActivityInput("act-cancel", null, "reason")); interceptor.terminateActivity( diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java index 13df137a3c..cf40c0aca9 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java @@ -3,11 +3,13 @@ import io.temporal.api.activity.v1.ActivityExecutionInfo; import io.temporal.api.enums.v1.ActivityExecutionStatus; import io.temporal.api.enums.v1.PendingActivityState; +import io.temporal.api.workflowservice.v1.DescribeActivityExecutionResponse; import io.temporal.common.Experimental; import io.temporal.common.Priority; import io.temporal.common.RetryOptions; import io.temporal.common.WorkerDeploymentVersion; import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.EncodedValues; import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.RetryOptionsUtils; @@ -27,45 +29,54 @@ @Experimental public final class ActivityExecutionDescription extends ActivityExecutionMetadata { - private final ActivityExecutionInfo info; + private final DescribeActivityExecutionResponse response; private final DataConverter dataConverter; - private final String namespace; public ActivityExecutionDescription( - ActivityExecutionInfo info, DataConverter dataConverter, String namespace) { + DescribeActivityExecutionResponse response, DataConverter dataConverter, String namespace) { super( null, - info.getActivityId(), - nullIfEmpty(info.getRunId()), - info.getActivityType().getName(), - info.hasCloseTime() ? ProtobufTimeUtils.toJavaInstant(info.getCloseTime()) : null, - info.hasExecutionDuration() - ? ProtobufTimeUtils.toJavaDuration(info.getExecutionDuration()) + response.getInfo().getActivityId(), + nullIfEmpty(response.getInfo().getRunId()), + response.getInfo().getActivityType().getName(), + response.getInfo().hasCloseTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getCloseTime()) : null, - info.hasScheduleTime() - ? ProtobufTimeUtils.toJavaInstant(info.getScheduleTime()) + response.getInfo().hasExecutionDuration() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getExecutionDuration()) + : null, + response.getInfo().hasScheduleTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getScheduleTime()) : Instant.EPOCH, - info.getStatus(), - info.getTaskQueue(), - SearchAttributesUtil.decodeTyped(info.getSearchAttributes())); - this.info = info; - this.dataConverter = dataConverter; - this.namespace = namespace; + response.getInfo().getStatus(), + response.getInfo().getTaskQueue(), + SearchAttributesUtil.decodeTyped(response.getInfo().getSearchAttributes())); + this.response = response; + this.dataConverter = + dataConverter.withContext( + new ActivitySerializationContext( + namespace, null, null, getActivityType(), getTaskQueue(), false)); } private static @Nullable String nullIfEmpty(String s) { return s == null || s.isEmpty() ? null : s; } + /** Underlying proto response. Exposed while the standalone activity surface is experimental. */ + @Nonnull + public DescribeActivityExecutionResponse getRawResponse() { + return response; + } + /** The raw protobuf info returned by the server for this activity execution. */ @Nonnull public ActivityExecutionInfo getRawInfo() { - return info; + return response.getInfo(); } /** Current attempt number (starts at 1). */ public int getAttempt() { - return info.getAttempt(); + return response.getInfo().getAttempt(); } /** @@ -74,83 +85,118 @@ public int getAttempt() { */ @Nullable public String getCanceledReason() { - String r = info.getCanceledReason(); + String r = response.getInfo().getCanceledReason(); return r.isEmpty() ? null : r; } /** Current or next retry interval. {@code null} if no retries are configured or allowed. */ @Nullable public Duration getCurrentRetryInterval() { - return info.hasCurrentRetryInterval() - ? ProtobufTimeUtils.toJavaDuration(info.getCurrentRetryInterval()) + return response.getInfo().hasCurrentRetryInterval() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getCurrentRetryInterval()) : null; } /** When the activity will time out (scheduled time + scheduleToCloseTimeout). */ @Nullable public Instant getExpirationTime() { - return info.hasExpirationTime() - ? ProtobufTimeUtils.toJavaInstant(info.getExpirationTime()) + return response.getInfo().hasExpirationTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getExpirationTime()) : null; } /** Maximum allowed time between heartbeats. */ @Nullable public Duration getHeartbeatTimeout() { - return info.hasHeartbeatTimeout() - ? ProtobufTimeUtils.toJavaDuration(info.getHeartbeatTimeout()) + return response.getInfo().hasHeartbeatTimeout() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getHeartbeatTimeout()) : null; } /** Time the last attempt completed (succeeded or failed). */ @Nullable public Instant getLastAttemptCompleteTime() { - return info.hasLastAttemptCompleteTime() - ? ProtobufTimeUtils.toJavaInstant(info.getLastAttemptCompleteTime()) + return response.getInfo().hasLastAttemptCompleteTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getLastAttemptCompleteTime()) : null; } /** Time the last heartbeat was recorded. */ @Nullable public Instant getLastHeartbeatTime() { - return info.hasLastHeartbeatTime() - ? ProtobufTimeUtils.toJavaInstant(info.getLastHeartbeatTime()) + return response.getInfo().hasLastHeartbeatTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getLastHeartbeatTime()) : null; } /** Time the last attempt was started. */ @Nullable public Instant getLastStartedTime() { - return info.hasLastStartedTime() - ? ProtobufTimeUtils.toJavaInstant(info.getLastStartedTime()) + return response.getInfo().hasLastStartedTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getLastStartedTime()) + : null; + } + + /** + * Time the first activity task was made available for dispatch. Computed as {@code schedule_time + * + start_delay}; equals {@code schedule_time} when no start delay is set. + */ + @Nullable + public Instant getExecutionTime() { + return response.getInfo().hasExecutionTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getExecutionTime()) + : null; + } + + /** + * Delay before the first activity task is made available for dispatch. Not applied to retry + * attempts. {@code null} if no start delay is set. + */ + @Nullable + public Duration getStartDelay() { + return response.getInfo().hasStartDelay() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getStartDelay()) : null; } + /** + * Whether a failure from a failed attempt is present. {@code false} when the activity has no + * failed attempt, and also when the description was requested without {@link + * DescribeActivityOptions.Builder#setIncludeLastFailure(boolean)}. + */ + public boolean hasLastFailure() { + return response.getInfo().hasLastFailure(); + } + /** Failure details from the last failed attempt. {@code null} if no failure has occurred. */ @Nullable - public Exception getLastFailure() { - return info.hasLastFailure() ? dataConverter.failureToException(info.getLastFailure()) : null; + public RuntimeException getLastFailure() { + return response.getInfo().hasLastFailure() + ? dataConverter.failureToException(response.getInfo().getLastFailure()) + : null; } /** Identity of the worker that last processed this activity. */ @Nullable public String getLastWorkerIdentity() { - String w = info.getLastWorkerIdentity(); + String w = response.getInfo().getLastWorkerIdentity(); return w.isEmpty() ? null : w; } /** Time when the next retry attempt will be scheduled. */ @Nullable public Instant getNextAttemptScheduleTime() { - return info.hasNextAttemptScheduleTime() - ? ProtobufTimeUtils.toJavaInstant(info.getNextAttemptScheduleTime()) + return response.getInfo().hasNextAttemptScheduleTime() + ? ProtobufTimeUtils.toJavaInstant(response.getInfo().getNextAttemptScheduleTime()) : null; } /** Retry policy for this activity. */ @Nullable public RetryOptions getRetryOptions() { - return info.hasRetryPolicy() ? RetryOptionsUtils.toRetryOptions(info.getRetryPolicy()) : null; + return response.getInfo().hasRetryPolicy() + ? RetryOptionsUtils.toRetryOptions(response.getInfo().getRetryPolicy()) + : null; } /** @@ -159,62 +205,119 @@ public RetryOptions getRetryOptions() { */ @Nonnull public PendingActivityState getRunState() { - return info.getRunState(); + return response.getInfo().getRunState(); } /** Total time the caller is willing to wait for the activity to complete, including retries. */ @Nullable public Duration getScheduleToCloseTimeout() { - return info.hasScheduleToCloseTimeout() - ? ProtobufTimeUtils.toJavaDuration(info.getScheduleToCloseTimeout()) + return response.getInfo().hasScheduleToCloseTimeout() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getScheduleToCloseTimeout()) : null; } /** Maximum time the task may wait in the task queue. */ @Nullable public Duration getScheduleToStartTimeout() { - return info.hasScheduleToStartTimeout() - ? ProtobufTimeUtils.toJavaDuration(info.getScheduleToStartTimeout()) + return response.getInfo().hasScheduleToStartTimeout() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getScheduleToStartTimeout()) : null; } /** Maximum time for a single attempt. */ @Nullable public Duration getStartToCloseTimeout() { - return info.hasStartToCloseTimeout() - ? ProtobufTimeUtils.toJavaDuration(info.getStartToCloseTimeout()) + return response.getInfo().hasStartToCloseTimeout() + ? ProtobufTimeUtils.toJavaDuration(response.getInfo().getStartToCloseTimeout()) : null; } - /** Whether heartbeat details were recorded for the last attempt. */ + /** + * Whether heartbeat details were recorded for the last attempt. {@code false} when the activity + * recorded none, and also when the description was requested without {@link + * DescribeActivityOptions.Builder#setIncludeHeartbeatDetails(boolean)}. + */ public boolean hasHeartbeatDetails() { - return info.hasHeartbeatDetails(); + return response.getInfo().hasHeartbeatDetails(); } /** - * Deserializes the last heartbeat details into the given type. Returns {@link Optional#empty()} - * if no heartbeat details are present. + * The details recorded by the last heartbeat, as lazily-decoded values. Empty (size 0) when no + * heartbeat details are present, either because none were recorded or because the description was + * requested without {@link DescribeActivityOptions.Builder#setIncludeHeartbeatDetails(boolean)}. + */ + public EncodedValues getHeartbeatDetails() { + return new EncodedValues(Optional.of(response.getInfo().getHeartbeatDetails()), dataConverter); + } + + /** + * Whether the activity's input is present. {@code false} unless the description was requested + * with {@link DescribeActivityOptions.Builder#setIncludeInput(boolean)}. + */ + public boolean hasInput() { + return response.hasInput(); + } + + /** + * The activity's input arguments, as lazily-decoded values, one per argument. Empty (size 0) when + * no input is present, either because the activity took no arguments or because the description + * was requested without {@link DescribeActivityOptions.Builder#setIncludeInput(boolean)}. + */ + public EncodedValues getInput() { + return new EncodedValues(Optional.of(response.getInput()), dataConverter); + } + + /** + * Whether the activity closed with a successful result. {@code false} while the activity is still + * running, when it closed with a failure, or when the description was requested without {@link + * DescribeActivityOptions.Builder#setIncludeOutcome(boolean)}. + */ + public boolean hasResult() { + return response.getOutcome().hasResult(); + } + + /** + * Deserializes the activity's success result. Returns {@link Optional#empty()} if no result is + * present (activity still running, closed with a failure, or {@code includeOutcome} was false). * - * @param valueType the class to deserialize the heartbeat details into + * @param valueType the class to deserialize the result into */ - public Optional getHeartbeatDetails(Class valueType) { - return getHeartbeatDetails(valueType, valueType); + public Optional getResult(Class valueType) { + return getResult(valueType, null); } /** - * Deserializes the last heartbeat details into the given generic type. Returns {@link - * Optional#empty()} if no heartbeat details are present. + * Deserializes the activity's success result into the given generic type. Returns {@link + * Optional#empty()} if no result is present. * - * @param valueType the class to deserialize the heartbeat details into + * @param valueType the class to deserialize the result into * @param genericType the generic type for deserialization; may equal {@code valueType} */ - public Optional getHeartbeatDetails(Class valueType, Type genericType) { - if (!info.hasHeartbeatDetails()) { + public Optional getResult(Class valueType, @Nullable Type genericType) { + if (!hasResult()) { return Optional.empty(); } return Optional.ofNullable( dataConverter.fromPayloads( - 0, Optional.of(info.getHeartbeatDetails()), valueType, genericType)); + 0, + Optional.of(response.getOutcome().getResult()), + valueType, + genericType != null ? genericType : valueType)); + } + + /** + * The failure the activity closed with, as an exception. {@code null} if the activity did not + * close with a failure or if {@code includeOutcome} was false on the describe call. + * + *

This is the terminal outcome; {@link #getLastFailure()} is the failure of the most recent + * attempt, which may be set while the activity is still retrying. + */ + @Nullable + public RuntimeException getOutcomeFailure() { + if (!response.getOutcome().hasFailure()) { + return null; + } + return dataConverter.failureToException(response.getOutcome().getFailure()); } /** @@ -223,20 +326,21 @@ public Optional getHeartbeatDetails(Class valueType, Type genericType) */ @Nullable public WorkerDeploymentVersion getWorkerDeploymentVersion() { - if (!info.hasLastDeploymentVersion()) { + if (!response.getInfo().hasLastDeploymentVersion()) { return null; } - io.temporal.api.deployment.v1.WorkerDeploymentVersion proto = info.getLastDeploymentVersion(); + io.temporal.api.deployment.v1.WorkerDeploymentVersion proto = + response.getInfo().getLastDeploymentVersion(); return new WorkerDeploymentVersion(proto.getDeploymentName(), proto.getBuildId()); } /** Priority hint for this activity. {@code null} if not set. */ @Nullable public Priority getPriority() { - if (!info.hasPriority()) { + if (!response.getInfo().hasPriority()) { return null; } - return ProtoConverters.fromProto(info.getPriority()); + return ProtoConverters.fromProto(response.getInfo().getPriority()); } /** @@ -245,14 +349,11 @@ public Priority getPriority() { */ @Nullable public String getStaticSummary() { - if (!info.hasUserMetadata() || !info.getUserMetadata().hasSummary()) { + if (!response.getInfo().getUserMetadata().hasSummary()) { return null; } - return dataConverter - .withContext( - new ActivitySerializationContext( - namespace, null, null, getActivityType(), getTaskQueue(), false)) - .fromPayload(info.getUserMetadata().getSummary(), String.class, String.class); + return dataConverter.fromPayload( + response.getInfo().getUserMetadata().getSummary(), String.class, String.class); } /** @@ -261,13 +362,10 @@ namespace, null, null, getActivityType(), getTaskQueue(), false)) */ @Nullable public String getStaticDetails() { - if (!info.hasUserMetadata() || !info.getUserMetadata().hasDetails()) { + if (!response.getInfo().getUserMetadata().hasDetails()) { return null; } - return dataConverter - .withContext( - new ActivitySerializationContext( - namespace, null, null, getActivityType(), getTaskQueue(), false)) - .fromPayload(info.getUserMetadata().getDetails(), String.class, String.class); + return dataConverter.fromPayload( + response.getInfo().getUserMetadata().getDetails(), String.class, String.class); } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/ActivityHandleImpl.java b/temporal-sdk/src/main/java/io/temporal/client/ActivityHandleImpl.java index 3144195d11..dd8864b60d 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/ActivityHandleImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/ActivityHandleImpl.java @@ -102,6 +102,11 @@ public ActivityExecutionDescription describe() { return delegate.describe(); } + @Override + public ActivityExecutionDescription describe(DescribeActivityOptions options) { + return delegate.describe(options); + } + @Override public void cancel() { delegate.cancel(); @@ -121,4 +126,44 @@ public void terminate() { public void terminate(@Nullable String reason) { delegate.terminate(reason); } + + @Override + public void pause() { + delegate.pause(); + } + + @Override + public void pause(PauseActivityOptions options) { + delegate.pause(options); + } + + @Override + public void unpause() { + delegate.unpause(); + } + + @Override + public void unpause(UnpauseActivityOptions options) { + delegate.unpause(options); + } + + @Override + public void reset() { + delegate.reset(); + } + + @Override + public void reset(ResetActivityOptions options) { + delegate.reset(options); + } + + @Override + public UpdateActivityOptions updateOptions(UpdateActivityOptions options) { + return delegate.updateOptions(options); + } + + @Override + public UpdateActivityOptions restoreOriginalOptions() { + return delegate.restoreOriginalOptions(); + } } diff --git a/temporal-sdk/src/main/java/io/temporal/client/DescribeActivityOptions.java b/temporal-sdk/src/main/java/io/temporal/client/DescribeActivityOptions.java new file mode 100644 index 0000000000..13e3093361 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/DescribeActivityOptions.java @@ -0,0 +1,145 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import java.util.Objects; + +/** + * Options for {@link UntypedActivityHandle#describe(DescribeActivityOptions)}. + * + *

Each flag opts in to a field on the description that carries a payload. Payloads can be + * arbitrarily large, so none are returned unless explicitly requested. An instance with no fields + * set describes the activity without any of them. + */ +@Experimental +public final class DescribeActivityOptions { + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(DescribeActivityOptions options) { + return new Builder(options); + } + + public static DescribeActivityOptions getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final DescribeActivityOptions DEFAULT_INSTANCE = + DescribeActivityOptions.newBuilder().build(); + + public static final class Builder { + private boolean includeInput; + private boolean includeOutcome; + private boolean includeHeartbeatDetails; + private boolean includeLastFailure; + + private Builder() {} + + private Builder(DescribeActivityOptions options) { + if (options == null) { + return; + } + this.includeInput = options.includeInput; + this.includeOutcome = options.includeOutcome; + this.includeHeartbeatDetails = options.includeHeartbeatDetails; + this.includeLastFailure = options.includeLastFailure; + } + + /** If set and the activity received input, the description includes the input. */ + public Builder setIncludeInput(boolean includeInput) { + this.includeInput = includeInput; + return this; + } + + /** If set and the activity is closed, the description includes the outcome. */ + public Builder setIncludeOutcome(boolean includeOutcome) { + this.includeOutcome = includeOutcome; + return this; + } + + /** + * If set and the activity recorded heartbeat details, the description includes the details of + * the last heartbeat. + */ + public Builder setIncludeHeartbeatDetails(boolean includeHeartbeatDetails) { + this.includeHeartbeatDetails = includeHeartbeatDetails; + return this; + } + + /** + * If set and the activity has a failed attempt, the description includes the failure of the + * last failed attempt. + */ + public Builder setIncludeLastFailure(boolean includeLastFailure) { + this.includeLastFailure = includeLastFailure; + return this; + } + + public DescribeActivityOptions build() { + return new DescribeActivityOptions(this); + } + } + + private final boolean includeInput; + private final boolean includeOutcome; + private final boolean includeHeartbeatDetails; + private final boolean includeLastFailure; + + private DescribeActivityOptions(Builder builder) { + this.includeInput = builder.includeInput; + this.includeOutcome = builder.includeOutcome; + this.includeHeartbeatDetails = builder.includeHeartbeatDetails; + this.includeLastFailure = builder.includeLastFailure; + } + + public Builder toBuilder() { + return new Builder(this); + } + + public boolean isIncludeInput() { + return includeInput; + } + + public boolean isIncludeOutcome() { + return includeOutcome; + } + + public boolean isIncludeHeartbeatDetails() { + return includeHeartbeatDetails; + } + + public boolean isIncludeLastFailure() { + return includeLastFailure; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + DescribeActivityOptions that = (DescribeActivityOptions) o; + return includeInput == that.includeInput + && includeOutcome == that.includeOutcome + && includeHeartbeatDetails == that.includeHeartbeatDetails + && includeLastFailure == that.includeLastFailure; + } + + @Override + public int hashCode() { + return Objects.hash(includeInput, includeOutcome, includeHeartbeatDetails, includeLastFailure); + } + + @Override + public String toString() { + return "DescribeActivityOptions{" + + "includeInput=" + + includeInput + + ", includeOutcome=" + + includeOutcome + + ", includeHeartbeatDetails=" + + includeHeartbeatDetails + + ", includeLastFailure=" + + includeLastFailure + + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/PauseActivityOptions.java b/temporal-sdk/src/main/java/io/temporal/client/PauseActivityOptions.java new file mode 100644 index 0000000000..f529839833 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/PauseActivityOptions.java @@ -0,0 +1,86 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Options for {@link UntypedActivityHandle#pause(PauseActivityOptions)}. + * + *

All fields are optional. An instance with no fields set pauses the activity with default + * behavior. + */ +@Experimental +public final class PauseActivityOptions { + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(PauseActivityOptions options) { + return new Builder(options); + } + + public static PauseActivityOptions getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final PauseActivityOptions DEFAULT_INSTANCE = + PauseActivityOptions.newBuilder().build(); + + public static final class Builder { + private @Nullable String reason; + + private Builder() {} + + private Builder(PauseActivityOptions options) { + if (options == null) { + return; + } + this.reason = options.reason; + } + + /** Human-readable reason for pausing, recorded on the server. */ + public Builder setReason(@Nullable String reason) { + this.reason = reason; + return this; + } + + public PauseActivityOptions build() { + return new PauseActivityOptions(this); + } + } + + private final @Nullable String reason; + + private PauseActivityOptions(Builder builder) { + this.reason = builder.reason; + } + + public Builder toBuilder() { + return new Builder(this); + } + + @Nullable + public String getReason() { + return reason; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + PauseActivityOptions that = (PauseActivityOptions) o; + return Objects.equals(reason, that.reason); + } + + @Override + public int hashCode() { + return Objects.hash(reason); + } + + @Override + public String toString() { + return "PauseActivityOptions{" + "reason='" + reason + "'" + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/ResetActivityOptions.java b/temporal-sdk/src/main/java/io/temporal/client/ResetActivityOptions.java new file mode 100644 index 0000000000..d959b2ce5e --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/ResetActivityOptions.java @@ -0,0 +1,152 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import java.time.Duration; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Options for {@link UntypedActivityHandle#reset(ResetActivityOptions)}. + * + *

All fields are optional. An instance with no fields set resets the activity with default + * behavior. + * + *

Reset does not clear recorded heartbeat details by default; set {@link + * Builder#setResetHeartbeat(boolean)} to additionally discard them. + */ +@Experimental +public final class ResetActivityOptions { + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(ResetActivityOptions options) { + return new Builder(options); + } + + public static ResetActivityOptions getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final ResetActivityOptions DEFAULT_INSTANCE = + ResetActivityOptions.newBuilder().build(); + + public static final class Builder { + private boolean keepPaused; + private @Nullable Duration jitter; + private boolean restoreOriginalOptions; + private boolean resetHeartbeat; + + private Builder() {} + + private Builder(ResetActivityOptions options) { + if (options == null) { + return; + } + this.keepPaused = options.keepPaused; + this.jitter = options.jitter; + this.restoreOriginalOptions = options.restoreOriginalOptions; + this.resetHeartbeat = options.resetHeartbeat; + } + + /** If set and the activity is paused, it will remain paused after the reset. */ + public Builder setKeepPaused(boolean keepPaused) { + this.keepPaused = keepPaused; + return this; + } + + /** + * If set and the activity is in backoff, the activity will start at a random time within the + * given jitter window (unless it is paused and {@link #setKeepPaused(boolean)} is set). + */ + public Builder setJitter(@Nullable Duration jitter) { + this.jitter = jitter; + return this; + } + + /** + * If set, the activity options are restored to the originals the activity was created with (the + * options recorded in the first schedule event). + * + *

This flag may be combined with other reset settings. + */ + public Builder setRestoreOriginalOptions(boolean restoreOriginalOptions) { + this.restoreOriginalOptions = restoreOriginalOptions; + return this; + } + + /** If set, reset additionally discards any persisted heartbeat details. */ + public Builder setResetHeartbeat(boolean resetHeartbeat) { + this.resetHeartbeat = resetHeartbeat; + return this; + } + + public ResetActivityOptions build() { + return new ResetActivityOptions(this); + } + } + + private final boolean keepPaused; + private final @Nullable Duration jitter; + private final boolean restoreOriginalOptions; + private final boolean resetHeartbeat; + + private ResetActivityOptions(Builder builder) { + this.keepPaused = builder.keepPaused; + this.jitter = builder.jitter; + this.restoreOriginalOptions = builder.restoreOriginalOptions; + this.resetHeartbeat = builder.resetHeartbeat; + } + + public Builder toBuilder() { + return new Builder(this); + } + + public boolean isKeepPaused() { + return keepPaused; + } + + @Nullable + public Duration getJitter() { + return jitter; + } + + public boolean isRestoreOriginalOptions() { + return restoreOriginalOptions; + } + + public boolean isResetHeartbeat() { + return resetHeartbeat; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + ResetActivityOptions that = (ResetActivityOptions) o; + return keepPaused == that.keepPaused + && restoreOriginalOptions == that.restoreOriginalOptions + && resetHeartbeat == that.resetHeartbeat + && Objects.equals(jitter, that.jitter); + } + + @Override + public int hashCode() { + return Objects.hash(keepPaused, jitter, restoreOriginalOptions, resetHeartbeat); + } + + @Override + public String toString() { + return "ResetActivityOptions{" + + "keepPaused=" + + keepPaused + + ", jitter=" + + jitter + + ", restoreOriginalOptions=" + + restoreOriginalOptions + + ", resetHeartbeat=" + + resetHeartbeat + + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/UnpauseActivityOptions.java b/temporal-sdk/src/main/java/io/temporal/client/UnpauseActivityOptions.java new file mode 100644 index 0000000000..c60da6a698 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/UnpauseActivityOptions.java @@ -0,0 +1,102 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import java.time.Duration; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Options for {@link UntypedActivityHandle#unpause(UnpauseActivityOptions)}. + * + *

All fields are optional. An instance with no fields set unpauses the activity with default + * behavior. + */ +@Experimental +public final class UnpauseActivityOptions { + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(UnpauseActivityOptions options) { + return new Builder(options); + } + + public static UnpauseActivityOptions getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final UnpauseActivityOptions DEFAULT_INSTANCE = + UnpauseActivityOptions.newBuilder().build(); + + public static final class Builder { + private @Nullable String reason; + private @Nullable Duration jitter; + + private Builder() {} + + private Builder(UnpauseActivityOptions options) { + if (options == null) { + return; + } + this.reason = options.reason; + this.jitter = options.jitter; + } + + /** Human-readable reason for unpausing. */ + public Builder setReason(@Nullable String reason) { + this.reason = reason; + return this; + } + + /** If set, the activity will resume at a random time within the given jitter window. */ + public Builder setJitter(@Nullable Duration jitter) { + this.jitter = jitter; + return this; + } + + public UnpauseActivityOptions build() { + return new UnpauseActivityOptions(this); + } + } + + private final @Nullable String reason; + private final @Nullable Duration jitter; + + private UnpauseActivityOptions(Builder builder) { + this.reason = builder.reason; + this.jitter = builder.jitter; + } + + public Builder toBuilder() { + return new Builder(this); + } + + @Nullable + public String getReason() { + return reason; + } + + @Nullable + public Duration getJitter() { + return jitter; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + UnpauseActivityOptions that = (UnpauseActivityOptions) o; + return Objects.equals(reason, that.reason) && Objects.equals(jitter, that.jitter); + } + + @Override + public int hashCode() { + return Objects.hash(reason, jitter); + } + + @Override + public String toString() { + return "UnpauseActivityOptions{" + "reason='" + reason + "', jitter=" + jitter + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/client/UntypedActivityHandle.java b/temporal-sdk/src/main/java/io/temporal/client/UntypedActivityHandle.java index 5e6bb12864..0fefbbf95e 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/UntypedActivityHandle.java +++ b/temporal-sdk/src/main/java/io/temporal/client/UntypedActivityHandle.java @@ -118,12 +118,22 @@ CompletableFuture getResultAsync( long timeout, TimeUnit unit, Class resultClass, @Nullable Type resultType); /** - * Describes the current state of the activity execution. + * Describes the current state of the activity execution, without any of the payload-bearing + * fields. Equivalent to {@code describe(DescribeActivityOptions.getDefaultInstance())}. * * @return detailed information about the activity */ ActivityExecutionDescription describe(); + /** + * Describes the current state of the activity execution. + * + * @param options which payload-bearing fields to include in the description. These are opt-in + * because they can be arbitrarily large. + * @return detailed information about the activity + */ + ActivityExecutionDescription describe(DescribeActivityOptions options); + /** * Requests cancellation of the activity. The activity will receive a cancellation via {@link * io.temporal.activity.ActivityExecutionContext#heartbeat(Object)}. @@ -146,4 +156,53 @@ CompletableFuture getResultAsync( * @param reason human-readable reason for termination, may be {@code null} */ void terminate(@Nullable String reason); + + /** + * Pauses the activity. A paused activity stops being dispatched to workers until it is unpaused. + */ + void pause(); + + /** + * Pauses the activity with the given options. + * + * @param options pause options (reason) + */ + void pause(PauseActivityOptions options); + + /** Unpauses the activity with default options, allowing it to be dispatched again. */ + void unpause(); + + /** + * Unpauses the activity with the given options. + * + * @param options unpause options (reset attempts, reset heartbeat, jitter, reason) + */ + void unpause(UnpauseActivityOptions options); + + /** Resets the activity with default options, scheduling a fresh attempt. */ + void reset(); + + /** + * Resets the activity with the given options. + * + * @param options reset options (reset heartbeat, keep paused, jitter, restore original options) + */ + void reset(ResetActivityOptions options); + + /** + * Updates the activity's options. Only the fields explicitly set in {@code options} are changed; + * a derived field mask leaves the rest untouched. To revert to the options the activity was + * created with, use {@link #restoreOriginalOptions()}. + * + * @param options the options to apply + * @return the activity options as resolved by the server after the update + */ + UpdateActivityOptions updateOptions(UpdateActivityOptions options); + + /** + * Restores the activity's options to the ones it was created with. + * + * @return the activity options as resolved by the server after the restore + */ + UpdateActivityOptions restoreOriginalOptions(); } diff --git a/temporal-sdk/src/main/java/io/temporal/client/UpdateActivityOptions.java b/temporal-sdk/src/main/java/io/temporal/client/UpdateActivityOptions.java new file mode 100644 index 0000000000..ba51df1fe5 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/client/UpdateActivityOptions.java @@ -0,0 +1,219 @@ +package io.temporal.client; + +import io.temporal.common.Experimental; +import io.temporal.common.Priority; +import io.temporal.common.RetryOptions; +import java.time.Duration; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Options for {@link UntypedActivityHandle#updateOptions(UpdateActivityOptions)}. + * + *

Only the fields that are explicitly set are sent to the server; a derived field mask ensures + * that unset fields are left unchanged (a partial update). + */ +@Experimental +public final class UpdateActivityOptions { + + public static Builder newBuilder() { + return new Builder(); + } + + public static Builder newBuilder(UpdateActivityOptions options) { + return new Builder(options); + } + + public static final class Builder { + private @Nullable String taskQueue; + private @Nullable Duration scheduleToCloseTimeout; + private @Nullable Duration scheduleToStartTimeout; + private @Nullable Duration startToCloseTimeout; + private @Nullable Duration heartbeatTimeout; + private @Nullable RetryOptions retryOptions; + private @Nullable Priority priority; + private @Nullable Duration startDelay; + + private Builder() {} + + private Builder(UpdateActivityOptions options) { + if (options == null) { + return; + } + this.taskQueue = options.taskQueue; + this.scheduleToCloseTimeout = options.scheduleToCloseTimeout; + this.scheduleToStartTimeout = options.scheduleToStartTimeout; + this.startToCloseTimeout = options.startToCloseTimeout; + this.heartbeatTimeout = options.heartbeatTimeout; + this.retryOptions = options.retryOptions; + this.priority = options.priority; + this.startDelay = options.startDelay; + } + + /** New task queue for the activity. */ + public Builder setTaskQueue(@Nullable String taskQueue) { + this.taskQueue = taskQueue; + return this; + } + + /** New schedule-to-close timeout. */ + public Builder setScheduleToCloseTimeout(@Nullable Duration scheduleToCloseTimeout) { + this.scheduleToCloseTimeout = scheduleToCloseTimeout; + return this; + } + + /** New schedule-to-start timeout. */ + public Builder setScheduleToStartTimeout(@Nullable Duration scheduleToStartTimeout) { + this.scheduleToStartTimeout = scheduleToStartTimeout; + return this; + } + + /** New start-to-close timeout. */ + public Builder setStartToCloseTimeout(@Nullable Duration startToCloseTimeout) { + this.startToCloseTimeout = startToCloseTimeout; + return this; + } + + /** New heartbeat timeout. */ + public Builder setHeartbeatTimeout(@Nullable Duration heartbeatTimeout) { + this.heartbeatTimeout = heartbeatTimeout; + return this; + } + + /** New retry policy. */ + public Builder setRetryOptions(@Nullable RetryOptions retryOptions) { + this.retryOptions = retryOptions; + return this; + } + + /** New priority. */ + public Builder setPriority(@Nullable Priority priority) { + this.priority = priority; + return this; + } + + /** New start delay for the first attempt. */ + public Builder setStartDelay(@Nullable Duration startDelay) { + this.startDelay = startDelay; + return this; + } + + public UpdateActivityOptions build() { + return new UpdateActivityOptions(this); + } + } + + private final @Nullable String taskQueue; + private final @Nullable Duration scheduleToCloseTimeout; + private final @Nullable Duration scheduleToStartTimeout; + private final @Nullable Duration startToCloseTimeout; + private final @Nullable Duration heartbeatTimeout; + private final @Nullable RetryOptions retryOptions; + private final @Nullable Priority priority; + private final @Nullable Duration startDelay; + + private UpdateActivityOptions(Builder builder) { + this.taskQueue = builder.taskQueue; + this.scheduleToCloseTimeout = builder.scheduleToCloseTimeout; + this.scheduleToStartTimeout = builder.scheduleToStartTimeout; + this.startToCloseTimeout = builder.startToCloseTimeout; + this.heartbeatTimeout = builder.heartbeatTimeout; + this.retryOptions = builder.retryOptions; + this.priority = builder.priority; + this.startDelay = builder.startDelay; + } + + public Builder toBuilder() { + return new Builder(this); + } + + @Nullable + public String getTaskQueue() { + return taskQueue; + } + + @Nullable + public Duration getScheduleToCloseTimeout() { + return scheduleToCloseTimeout; + } + + @Nullable + public Duration getScheduleToStartTimeout() { + return scheduleToStartTimeout; + } + + @Nullable + public Duration getStartToCloseTimeout() { + return startToCloseTimeout; + } + + @Nullable + public Duration getHeartbeatTimeout() { + return heartbeatTimeout; + } + + @Nullable + public RetryOptions getRetryOptions() { + return retryOptions; + } + + @Nullable + public Priority getPriority() { + return priority; + } + + @Nullable + public Duration getStartDelay() { + return startDelay; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + UpdateActivityOptions that = (UpdateActivityOptions) o; + return Objects.equals(taskQueue, that.taskQueue) + && Objects.equals(scheduleToCloseTimeout, that.scheduleToCloseTimeout) + && Objects.equals(scheduleToStartTimeout, that.scheduleToStartTimeout) + && Objects.equals(startToCloseTimeout, that.startToCloseTimeout) + && Objects.equals(heartbeatTimeout, that.heartbeatTimeout) + && Objects.equals(retryOptions, that.retryOptions) + && Objects.equals(priority, that.priority) + && Objects.equals(startDelay, that.startDelay); + } + + @Override + public int hashCode() { + return Objects.hash( + taskQueue, + scheduleToCloseTimeout, + scheduleToStartTimeout, + startToCloseTimeout, + heartbeatTimeout, + retryOptions, + priority, + startDelay); + } + + @Override + public String toString() { + return "UpdateActivityOptions{" + + "taskQueue='" + + taskQueue + + "', scheduleToCloseTimeout=" + + scheduleToCloseTimeout + + ", scheduleToStartTimeout=" + + scheduleToStartTimeout + + ", startToCloseTimeout=" + + startToCloseTimeout + + ", heartbeatTimeout=" + + heartbeatTimeout + + ", retryOptions=" + + retryOptions + + ", priority=" + + priority + + ", startDelay=" + + startDelay + + '}'; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java index 16a34dc285..9b167be3bf 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java @@ -1,11 +1,18 @@ package io.temporal.common.interceptors; +import com.google.protobuf.FieldMask; +import io.temporal.api.activity.v1.ActivityOptions; import io.temporal.client.ActivityAlreadyStartedException; import io.temporal.client.ActivityExecutionCount; import io.temporal.client.ActivityExecutionDescription; import io.temporal.client.ActivityExecutionMetadata; import io.temporal.client.ActivityFailedException; +import io.temporal.client.DescribeActivityOptions; +import io.temporal.client.PauseActivityOptions; +import io.temporal.client.ResetActivityOptions; import io.temporal.client.StartActivityOptions; +import io.temporal.client.UnpauseActivityOptions; +import io.temporal.client.UpdateActivityOptions; import io.temporal.common.Experimental; import java.lang.reflect.Type; import java.util.List; @@ -80,6 +87,42 @@ GetActivityResultOutput getActivityResult(GetActivityResultInput input */ TerminateActivityOutput terminateActivity(TerminateActivityInput input); + /** + * Pauses a running standalone activity. A paused activity stops being dispatched to workers until + * it is unpaused. + * + * @param input activity ID, optional run ID, and optional human-readable reason + * @return an empty output object (reserved for future use) + */ + PauseActivityOutput pauseActivity(PauseActivityInput input); + + /** + * Unpauses a previously paused standalone activity, optionally resetting its attempt counter and + * heartbeat details. + * + * @param input activity ID, optional run ID, and unpause options + * @return an empty output object (reserved for future use) + */ + UnpauseActivityOutput unpauseActivity(UnpauseActivityInput input); + + /** + * Resets a standalone activity, scheduling a fresh attempt. + * + * @param input activity ID, optional run ID, and reset options + * @return an empty output object (reserved for future use) + */ + ResetActivityOutput resetActivity(ResetActivityInput input); + + /** + * Updates the options of a standalone activity. The {@code updateMask} controls which fields of + * {@code activityOptions} are applied; alternatively {@code restoreOriginal} reverts the options + * to the values the activity was created with. + * + * @param input activity ID, optional run ID, options, update mask, and restore flag + * @return output carrying the activity options as resolved by the server after the update + */ + UpdateActivityOptionsOutput updateActivityOptions(UpdateActivityOptionsInput input); + /** * Returns a lazy {@link java.util.stream.Stream} of activity execution metadata matching the * Visibility query in {@code input}. Pages are fetched from the server on demand as the stream is @@ -250,10 +293,13 @@ public R getResult() { final class DescribeActivityInput { private final String id; private final @Nullable String runId; + private final DescribeActivityOptions options; - public DescribeActivityInput(String id, @Nullable String runId) { + public DescribeActivityInput( + String id, @Nullable String runId, DescribeActivityOptions options) { this.id = id; this.runId = runId; + this.options = options; } public String getId() { @@ -264,6 +310,10 @@ public String getId() { public String getRunId() { return runId; } + + public DescribeActivityOptions getOptions() { + return options; + } } @Experimental @@ -339,6 +389,150 @@ public String getReason() { @Experimental final class TerminateActivityOutput {} + @Experimental + final class PauseActivityInput { + private final String id; + private final @Nullable String runId; + private final PauseActivityOptions options; + + public PauseActivityInput(String id, @Nullable String runId, PauseActivityOptions options) { + this.id = id; + this.runId = runId; + this.options = options; + } + + public String getId() { + return id; + } + + @Nullable + public String getRunId() { + return runId; + } + + public PauseActivityOptions getOptions() { + return options; + } + } + + @Experimental + final class PauseActivityOutput {} + + @Experimental + final class UnpauseActivityInput { + private final String id; + private final @Nullable String runId; + private final UnpauseActivityOptions options; + + public UnpauseActivityInput(String id, @Nullable String runId, UnpauseActivityOptions options) { + this.id = id; + this.runId = runId; + this.options = options; + } + + public String getId() { + return id; + } + + @Nullable + public String getRunId() { + return runId; + } + + public UnpauseActivityOptions getOptions() { + return options; + } + } + + @Experimental + final class UnpauseActivityOutput {} + + @Experimental + final class ResetActivityInput { + private final String id; + private final @Nullable String runId; + private final ResetActivityOptions options; + + public ResetActivityInput(String id, @Nullable String runId, ResetActivityOptions options) { + this.id = id; + this.runId = runId; + this.options = options; + } + + public String getId() { + return id; + } + + @Nullable + public String getRunId() { + return runId; + } + + public ResetActivityOptions getOptions() { + return options; + } + } + + @Experimental + final class ResetActivityOutput {} + + @Experimental + final class UpdateActivityOptionsInput { + private final String id; + private final @Nullable String runId; + private final ActivityOptions activityOptions; + private final FieldMask updateMask; + private final boolean restoreOriginal; + + public UpdateActivityOptionsInput( + String id, + @Nullable String runId, + ActivityOptions activityOptions, + FieldMask updateMask, + boolean restoreOriginal) { + this.id = id; + this.runId = runId; + this.activityOptions = activityOptions; + this.updateMask = updateMask; + this.restoreOriginal = restoreOriginal; + } + + public String getId() { + return id; + } + + @Nullable + public String getRunId() { + return runId; + } + + public ActivityOptions getActivityOptions() { + return activityOptions; + } + + public FieldMask getUpdateMask() { + return updateMask; + } + + public boolean isRestoreOriginal() { + return restoreOriginal; + } + } + + @Experimental + final class UpdateActivityOptionsOutput { + private final UpdateActivityOptions options; + + public UpdateActivityOptionsOutput(UpdateActivityOptions options) { + this.options = options; + } + + /** The activity options as resolved by the server after the update. */ + public UpdateActivityOptions getOptions() { + return options; + } + } + @Experimental final class ListActivitiesInput { private final String query; diff --git a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBase.java b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBase.java index e8b99f5b9f..25256f7ce4 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBase.java +++ b/temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBase.java @@ -1,9 +1,11 @@ package io.temporal.common.interceptors; +import io.temporal.common.Experimental; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeoutException; /** Convenience base class for {@link ActivityClientCallsInterceptor} implementations. */ +@Experimental public class ActivityClientCallsInterceptorBase implements ActivityClientCallsInterceptor { private final ActivityClientCallsInterceptor next; @@ -44,6 +46,26 @@ public TerminateActivityOutput terminateActivity(TerminateActivityInput input) { return next.terminateActivity(input); } + @Override + public PauseActivityOutput pauseActivity(PauseActivityInput input) { + return next.pauseActivity(input); + } + + @Override + public UnpauseActivityOutput unpauseActivity(UnpauseActivityInput input) { + return next.unpauseActivity(input); + } + + @Override + public ResetActivityOutput resetActivity(ResetActivityInput input) { + return next.resetActivity(input); + } + + @Override + public UpdateActivityOptionsOutput updateActivityOptions(UpdateActivityOptionsInput input) { + return next.updateActivityOptions(input); + } + @Override public ListActivitiesOutput listActivities(ListActivitiesInput input) { return next.listActivities(input); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHandleImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHandleImpl.java index 77ecddcb4f..85103c8397 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHandleImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHandleImpl.java @@ -1,9 +1,23 @@ package io.temporal.internal.client; +import static io.temporal.internal.common.RetryOptionsUtils.toRetryPolicy; + +import com.google.protobuf.FieldMask; +import io.temporal.api.activity.v1.ActivityOptions; +import io.temporal.api.taskqueue.v1.TaskQueue; import io.temporal.client.ActivityExecutionDescription; +import io.temporal.client.DescribeActivityOptions; +import io.temporal.client.PauseActivityOptions; +import io.temporal.client.ResetActivityOptions; +import io.temporal.client.UnpauseActivityOptions; import io.temporal.client.UntypedActivityHandle; +import io.temporal.client.UpdateActivityOptions; import io.temporal.common.interceptors.ActivityClientCallsInterceptor; +import io.temporal.internal.common.ProtoConverters; +import io.temporal.internal.common.ProtobufTimeUtils; import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -102,9 +116,15 @@ public CompletableFuture getResultAsync( @Override public ActivityExecutionDescription describe() { + return describe(DescribeActivityOptions.getDefaultInstance()); + } + + @Override + public ActivityExecutionDescription describe(DescribeActivityOptions options) { return clientCallsInterceptor .describeActivity( - new ActivityClientCallsInterceptor.DescribeActivityInput(activityId, activityRunId)) + new ActivityClientCallsInterceptor.DescribeActivityInput( + activityId, activityRunId, options)) .getDescription(); } @@ -130,4 +150,103 @@ public void terminate(@Nullable String reason) { new ActivityClientCallsInterceptor.TerminateActivityInput( activityId, activityRunId, reason)); } + + @Override + public void pause() { + pause(PauseActivityOptions.getDefaultInstance()); + } + + @Override + public void pause(PauseActivityOptions options) { + clientCallsInterceptor.pauseActivity( + new ActivityClientCallsInterceptor.PauseActivityInput(activityId, activityRunId, options)); + } + + @Override + public void unpause() { + unpause(UnpauseActivityOptions.getDefaultInstance()); + } + + @Override + public void unpause(UnpauseActivityOptions options) { + clientCallsInterceptor.unpauseActivity( + new ActivityClientCallsInterceptor.UnpauseActivityInput( + activityId, activityRunId, options)); + } + + @Override + public void reset() { + reset(ResetActivityOptions.getDefaultInstance()); + } + + @Override + public void reset(ResetActivityOptions options) { + clientCallsInterceptor.resetActivity( + new ActivityClientCallsInterceptor.ResetActivityInput(activityId, activityRunId, options)); + } + + @Override + public UpdateActivityOptions updateOptions(UpdateActivityOptions options) { + ActivityOptions.Builder activityOptions = ActivityOptions.newBuilder(); + List maskPaths = new ArrayList<>(); + + if (options.getTaskQueue() != null) { + activityOptions.setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build()); + maskPaths.add("task_queue.name"); + } + if (options.getScheduleToCloseTimeout() != null) { + activityOptions.setScheduleToCloseTimeout( + ProtobufTimeUtils.toProtoDuration(options.getScheduleToCloseTimeout())); + maskPaths.add("schedule_to_close_timeout"); + } + if (options.getScheduleToStartTimeout() != null) { + activityOptions.setScheduleToStartTimeout( + ProtobufTimeUtils.toProtoDuration(options.getScheduleToStartTimeout())); + maskPaths.add("schedule_to_start_timeout"); + } + if (options.getStartToCloseTimeout() != null) { + activityOptions.setStartToCloseTimeout( + ProtobufTimeUtils.toProtoDuration(options.getStartToCloseTimeout())); + maskPaths.add("start_to_close_timeout"); + } + if (options.getHeartbeatTimeout() != null) { + activityOptions.setHeartbeatTimeout( + ProtobufTimeUtils.toProtoDuration(options.getHeartbeatTimeout())); + maskPaths.add("heartbeat_timeout"); + } + if (options.getRetryOptions() != null) { + activityOptions.setRetryPolicy(toRetryPolicy(options.getRetryOptions())); + maskPaths.add("retry_policy"); + } + if (options.getPriority() != null) { + activityOptions.setPriority(ProtoConverters.toProto(options.getPriority())); + maskPaths.add("priority"); + } + if (options.getStartDelay() != null) { + activityOptions.setStartDelay(ProtobufTimeUtils.toProtoDuration(options.getStartDelay())); + maskPaths.add("start_delay"); + } + + FieldMask updateMask = FieldMask.newBuilder().addAllPaths(maskPaths).build(); + + ActivityClientCallsInterceptor.UpdateActivityOptionsOutput output = + clientCallsInterceptor.updateActivityOptions( + new ActivityClientCallsInterceptor.UpdateActivityOptionsInput( + activityId, activityRunId, activityOptions.build(), updateMask, false)); + + return output.getOptions(); + } + + @Override + public UpdateActivityOptions restoreOriginalOptions() { + ActivityClientCallsInterceptor.UpdateActivityOptionsOutput output = + clientCallsInterceptor.updateActivityOptions( + new ActivityClientCallsInterceptor.UpdateActivityOptionsInput( + activityId, + activityRunId, + ActivityOptions.getDefaultInstance(), + FieldMask.getDefaultInstance(), + true)); + return output.getOptions(); + } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 3d4b1155d3..cb5f3209c4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -9,6 +9,7 @@ import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.temporal.api.activity.v1.ActivityExecutionOutcome; +import io.temporal.api.activity.v1.ActivityOptions; import io.temporal.api.common.v1.ActivityType; import io.temporal.api.common.v1.Callback; import io.temporal.api.common.v1.Link; @@ -25,6 +26,7 @@ import io.temporal.internal.common.InternalUtils; import io.temporal.internal.common.ProtoConverters; import io.temporal.internal.common.ProtobufTimeUtils; +import io.temporal.internal.common.RetryOptionsUtils; import io.temporal.internal.common.SearchAttributesUtil; import io.temporal.internal.nexus.CurrentNexusOperationContext; import io.temporal.internal.nexus.InternalNexusOperationContext; @@ -339,14 +341,19 @@ public DescribeActivityOutput describeActivity(DescribeActivityInput input) { DescribeActivityExecutionRequest.Builder req = DescribeActivityExecutionRequest.newBuilder() .setNamespace(clientOptions.getNamespace()) - .setActivityId(input.getId()); + .setActivityId(input.getId()) + .setIncludeInput(input.getOptions().isIncludeInput()) + .setIncludeOutcome(input.getOptions().isIncludeOutcome()) + .setIncludeHeartbeatDetails(input.getOptions().isIncludeHeartbeatDetails()) + .setIncludeLastFailure(input.getOptions().isIncludeLastFailure()); if (input.getRunId() != null) { req.setRunId(input.getRunId()); } - DescribeActivityExecutionResponse response = genericClient.describeActivity(req.build()); + DescribeActivityExecutionResponse response = + stripUnrequestedPayloads(genericClient.describeActivity(req.build()), input.getOptions()); return new DescribeActivityOutput( new ActivityExecutionDescription( - response.getInfo(), clientOptions.getDataConverter(), clientOptions.getNamespace())); + response, clientOptions.getDataConverter(), clientOptions.getNamespace())); } @Override @@ -385,6 +392,87 @@ public TerminateActivityOutput terminateActivity(TerminateActivityInput input) { return new TerminateActivityOutput(); } + @Override + public PauseActivityOutput pauseActivity(PauseActivityInput input) { + PauseActivityExecutionRequest.Builder req = + PauseActivityExecutionRequest.newBuilder() + .setNamespace(clientOptions.getNamespace()) + .setIdentity(clientOptions.getIdentity()) + .setRequestId(UUID.randomUUID().toString()) + .setActivityId(input.getId()); + if (input.getRunId() != null) { + req.setRunId(input.getRunId()); + } + if (input.getOptions().getReason() != null) { + req.setReason(input.getOptions().getReason()); + } + genericClient.pauseActivity(req.build()); + return new PauseActivityOutput(); + } + + @Override + public UnpauseActivityOutput unpauseActivity(UnpauseActivityInput input) { + UnpauseActivityExecutionRequest.Builder req = + UnpauseActivityExecutionRequest.newBuilder() + .setNamespace(clientOptions.getNamespace()) + .setIdentity(clientOptions.getIdentity()) + .setActivityId(input.getId()) + .setRequestId(UUID.randomUUID().toString()); + if (input.getRunId() != null) { + req.setRunId(input.getRunId()); + } + if (input.getOptions().getReason() != null) { + req.setReason(input.getOptions().getReason()); + } + if (input.getOptions().getJitter() != null) { + req.setJitter(ProtobufTimeUtils.toProtoDuration(input.getOptions().getJitter())); + } + genericClient.unpauseActivity(req.build()); + return new UnpauseActivityOutput(); + } + + @Override + public ResetActivityOutput resetActivity(ResetActivityInput input) { + ResetActivityExecutionRequest.Builder req = + ResetActivityExecutionRequest.newBuilder() + .setNamespace(clientOptions.getNamespace()) + .setIdentity(clientOptions.getIdentity()) + .setActivityId(input.getId()) + .setRequestId(UUID.randomUUID().toString()) + .setKeepPaused(input.getOptions().isKeepPaused()) + .setRestoreOriginalOptions(input.getOptions().isRestoreOriginalOptions()) + .setResetHeartbeat(input.getOptions().isResetHeartbeat()); + if (input.getRunId() != null) { + req.setRunId(input.getRunId()); + } + if (input.getOptions().getJitter() != null) { + req.setJitter(ProtobufTimeUtils.toProtoDuration(input.getOptions().getJitter())); + } + genericClient.resetActivity(req.build()); + return new ResetActivityOutput(); + } + + @Override + public UpdateActivityOptionsOutput updateActivityOptions(UpdateActivityOptionsInput input) { + UpdateActivityExecutionOptionsRequest.Builder req = + UpdateActivityExecutionOptionsRequest.newBuilder() + .setNamespace(clientOptions.getNamespace()) + .setIdentity(clientOptions.getIdentity()) + .setActivityId(input.getId()) + .setRequestId(UUID.randomUUID().toString()); + if (input.getRunId() != null) { + req.setRunId(input.getRunId()); + } + if (input.isRestoreOriginal()) { + req.setRestoreOriginal(true); + } else { + req.setActivityOptions(input.getActivityOptions()).setUpdateMask(input.getUpdateMask()); + } + UpdateActivityExecutionOptionsResponse response = + genericClient.updateActivityOptions(req.build()); + return new UpdateActivityOptionsOutput(toUpdateActivityOptions(response.getActivityOptions())); + } + @Override public ListActivitiesOutput listActivities(ListActivitiesInput input) { ListActivityExecutionIterator iterator = @@ -410,4 +498,65 @@ public CountActivitiesOutput countActivities(CountActivitiesInput input) { CountActivityExecutionsResponse resp = genericClient.countActivities(req.build()); return new CountActivitiesOutput(new ActivityExecutionCount(resp)); } + + /** + * Clears payload-bearing fields the caller did not ask for, in case an older or buggy server sent + * them anyway. + */ + private static DescribeActivityExecutionResponse stripUnrequestedPayloads( + DescribeActivityExecutionResponse response, DescribeActivityOptions options) { + if (options.isIncludeInput() + && options.isIncludeOutcome() + && options.isIncludeHeartbeatDetails() + && options.isIncludeLastFailure()) { + return response; + } + DescribeActivityExecutionResponse.Builder builder = response.toBuilder(); + if (!options.isIncludeInput()) { + builder.clearInput(); + } + if (!options.isIncludeOutcome()) { + builder.clearOutcome(); + } + if (!options.isIncludeHeartbeatDetails()) { + builder.getInfoBuilder().clearHeartbeatDetails(); + } + if (!options.isIncludeLastFailure()) { + builder.getInfoBuilder().clearLastFailure(); + } + return builder.build(); + } + + /** Converts the server's resolved activity options into the public options type. */ + private static UpdateActivityOptions toUpdateActivityOptions(ActivityOptions proto) { + UpdateActivityOptions.Builder builder = UpdateActivityOptions.newBuilder(); + if (proto.hasTaskQueue()) { + builder.setTaskQueue(proto.getTaskQueue().getName()); + } + if (proto.hasScheduleToCloseTimeout()) { + builder.setScheduleToCloseTimeout( + ProtobufTimeUtils.toJavaDuration(proto.getScheduleToCloseTimeout())); + } + if (proto.hasScheduleToStartTimeout()) { + builder.setScheduleToStartTimeout( + ProtobufTimeUtils.toJavaDuration(proto.getScheduleToStartTimeout())); + } + if (proto.hasStartToCloseTimeout()) { + builder.setStartToCloseTimeout( + ProtobufTimeUtils.toJavaDuration(proto.getStartToCloseTimeout())); + } + if (proto.hasHeartbeatTimeout()) { + builder.setHeartbeatTimeout(ProtobufTimeUtils.toJavaDuration(proto.getHeartbeatTimeout())); + } + if (proto.hasRetryPolicy()) { + builder.setRetryOptions(RetryOptionsUtils.toRetryOptions(proto.getRetryPolicy())); + } + if (proto.hasPriority()) { + builder.setPriority(ProtoConverters.fromProto(proto.getPriority())); + } + if (proto.hasStartDelay()) { + builder.setStartDelay(ProtobufTimeUtils.toJavaDuration(proto.getStartDelay())); + } + return builder.build(); + } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClient.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClient.java index 23932104fe..94eb7c1ce5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClient.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClient.java @@ -122,6 +122,19 @@ CompletableFuture pollActivityAsync( @Experimental void terminateActivity(TerminateActivityExecutionRequest request); + @Experimental + void pauseActivity(PauseActivityExecutionRequest request); + + @Experimental + void unpauseActivity(UnpauseActivityExecutionRequest request); + + @Experimental + void resetActivity(ResetActivityExecutionRequest request); + + @Experimental + UpdateActivityExecutionOptionsResponse updateActivityOptions( + UpdateActivityExecutionOptionsRequest request); + @Experimental ListActivityExecutionsResponse listActivities(ListActivityExecutionsRequest request); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java index cee4ffc893..ec0f1b7076 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/GenericWorkflowClientImpl.java @@ -641,6 +641,51 @@ public void terminateActivity(TerminateActivityExecutionRequest request) { grpcRetryerOptions); } + @Override + public void pauseActivity(PauseActivityExecutionRequest request) { + grpcRetryer.retry( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .pauseActivityExecution(request), + grpcRetryerOptions); + } + + @Override + public void unpauseActivity(UnpauseActivityExecutionRequest request) { + grpcRetryer.retry( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .unpauseActivityExecution(request), + grpcRetryerOptions); + } + + @Override + public void resetActivity(ResetActivityExecutionRequest request) { + grpcRetryer.retry( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .resetActivityExecution(request), + grpcRetryerOptions); + } + + @Override + public UpdateActivityExecutionOptionsResponse updateActivityOptions( + UpdateActivityExecutionOptionsRequest request) { + return grpcRetryer.retryWithResult( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .updateActivityExecutionOptions(request), + grpcRetryerOptions); + } + @Override public ListActivityExecutionsResponse listActivities(ListActivityExecutionsRequest request) { return grpcRetryer.retryWithResult( diff --git a/temporal-sdk/src/test/java/io/temporal/client/ActivityExecutionDescriptionTest.java b/temporal-sdk/src/test/java/io/temporal/client/ActivityExecutionDescriptionTest.java index 024d8b1890..6c965c45bc 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/ActivityExecutionDescriptionTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/ActivityExecutionDescriptionTest.java @@ -4,19 +4,21 @@ import com.google.common.reflect.TypeToken; import io.temporal.api.activity.v1.ActivityExecutionInfo; +import io.temporal.api.activity.v1.ActivityExecutionOutcome; import io.temporal.api.common.v1.ActivityType; import io.temporal.api.common.v1.Payloads; import io.temporal.api.enums.v1.ActivityExecutionStatus; +import io.temporal.api.workflowservice.v1.DescribeActivityExecutionResponse; import io.temporal.common.Priority; import io.temporal.common.WorkerDeploymentVersion; import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.failure.ApplicationFailure; import io.temporal.internal.common.ProtobufTimeUtils; import java.lang.reflect.Type; import java.time.Instant; import java.util.Arrays; import java.util.List; -import java.util.Optional; import org.junit.Test; public class ActivityExecutionDescriptionTest { @@ -35,26 +37,31 @@ private ActivityExecutionInfo buildInfo(String activityId, String runId) { .build(); } + private ActivityExecutionDescription describe(ActivityExecutionInfo info) { + return describe(DescribeActivityExecutionResponse.newBuilder().setInfo(info).build()); + } + + private ActivityExecutionDescription describe(DescribeActivityExecutionResponse response) { + return new ActivityExecutionDescription(response, CONVERTER, "test-ns"); + } + @Test public void testNullRunIdWhenEmpty() { - ActivityExecutionDescription desc = - new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(buildInfo("act-id", "")); assertNull(desc.getActivityRunId()); } @Test public void testScheduledTime() { - ActivityExecutionDescription desc = - new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(buildInfo("act-id", "")); assertEquals(Instant.ofEpochMilli(1000), desc.getScheduledTime()); } @Test public void testHasHeartbeatDetailsAbsent() { - ActivityExecutionDescription desc = - new ActivityExecutionDescription(buildInfo("id", "run"), CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(buildInfo("id", "run")); assertFalse(desc.hasHeartbeatDetails()); - assertFalse(desc.getHeartbeatDetails(String.class).isPresent()); + assertEquals(0, desc.getHeartbeatDetails().getSize()); } @Test @@ -62,13 +69,11 @@ public void testGetHeartbeatDetailsPresent() { Payloads encoded = CONVERTER.toPayloads("hello-heartbeat").get(); ActivityExecutionInfo info = buildInfo("id", "run").toBuilder().setHeartbeatDetails(encoded).build(); - ActivityExecutionDescription desc = - new ActivityExecutionDescription(info, CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(info); assertTrue(desc.hasHeartbeatDetails()); - Optional result = desc.getHeartbeatDetails(String.class); - assertTrue(result.isPresent()); - assertEquals("hello-heartbeat", result.get()); + assertEquals(1, desc.getHeartbeatDetails().getSize()); + assertEquals("hello-heartbeat", desc.getHeartbeatDetails().get(0, String.class)); } @Test @@ -78,14 +83,13 @@ public void testGetHeartbeatDetailsWithExplicitGenericType() { Payloads encoded = CONVERTER.toPayloads(original).get(); ActivityExecutionInfo info = buildInfo("id", "run").toBuilder().setHeartbeatDetails(encoded).build(); - ActivityExecutionDescription desc = - new ActivityExecutionDescription(info, CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(info); Type genericType = new TypeToken>() {}.getType(); Class> listClass = (Class>) (Class) List.class; - Optional> result = desc.getHeartbeatDetails(listClass, genericType); - assertTrue(result.isPresent()); - assertEquals(Arrays.asList("one", "two", "three"), result.get()); + assertEquals( + Arrays.asList("one", "two", "three"), + desc.getHeartbeatDetails().get(0, listClass, genericType)); } @Test @@ -97,8 +101,7 @@ public void testGetWorkerDeploymentVersionPresent() { .build(); ActivityExecutionInfo info = buildInfo("id", "run").toBuilder().setLastDeploymentVersion(protoVersion).build(); - ActivityExecutionDescription desc = - new ActivityExecutionDescription(info, CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(info); WorkerDeploymentVersion version = desc.getWorkerDeploymentVersion(); assertNotNull(version); @@ -106,14 +109,104 @@ public void testGetWorkerDeploymentVersionPresent() { assertEquals("build-42", version.getBuildId()); } + @Test + public void testInputAbsentUnlessRequested() { + ActivityExecutionDescription desc = describe(buildInfo("id", "run")); + assertFalse(desc.hasInput()); + assertEquals(0, desc.getInput().getSize()); + } + + @Test + public void testGetInputPresent() { + DescribeActivityExecutionResponse response = + DescribeActivityExecutionResponse.newBuilder() + .setInfo(buildInfo("id", "run")) + .setInput(CONVERTER.toPayloads("hello-input").get()) + .build(); + ActivityExecutionDescription desc = describe(response); + + assertTrue(desc.hasInput()); + assertEquals(1, desc.getInput().getSize()); + assertEquals("hello-input", desc.getInput().get(0, String.class)); + } + + @Test + public void testGetInputDecodesEveryArgument() { + DescribeActivityExecutionResponse response = + DescribeActivityExecutionResponse.newBuilder() + .setInfo(buildInfo("id", "run")) + .setInput(CONVERTER.toPayloads("first", 42).get()) + .build(); + ActivityExecutionDescription desc = describe(response); + + assertEquals(2, desc.getInput().getSize()); + assertEquals("first", desc.getInput().get(0, String.class)); + assertEquals(Integer.valueOf(42), desc.getInput().get(1, Integer.class)); + } + + @Test + public void testInputEmptyWhenInputAbsent() { + ActivityExecutionDescription desc = describe(buildInfo("id", "run")); + assertEquals(0, desc.getInput().getSize()); + } + + @Test + public void testOutcomeAbsentUnlessRequested() { + ActivityExecutionDescription desc = describe(buildInfo("id", "run")); + assertFalse(desc.hasResult()); + assertFalse(desc.getResult(String.class).isPresent()); + assertNull(desc.getOutcomeFailure()); + } + + @Test + public void testGetResultPresentOnSuccessfulOutcome() { + DescribeActivityExecutionResponse response = + DescribeActivityExecutionResponse.newBuilder() + .setInfo(buildInfo("id", "run")) + .setOutcome( + ActivityExecutionOutcome.newBuilder() + .setResult(CONVERTER.toPayloads("hello-result").get()) + .build()) + .build(); + ActivityExecutionDescription desc = describe(response); + + assertTrue(desc.hasResult()); + assertEquals("hello-result", desc.getResult(String.class).orElse(null)); + // A successful outcome has no failure arm. + assertNull(desc.getOutcomeFailure()); + } + + @Test + public void testGetFailurePresentOnFailedOutcome() { + DescribeActivityExecutionResponse response = + DescribeActivityExecutionResponse.newBuilder() + .setInfo(buildInfo("id", "run")) + .setOutcome( + ActivityExecutionOutcome.newBuilder() + .setFailure( + CONVERTER.exceptionToFailure( + ApplicationFailure.newFailure("boom", "test-type"))) + .build()) + .build(); + ActivityExecutionDescription desc = describe(response); + + // The failure arm is populated, so there is no result to read. + assertFalse(desc.hasResult()); + assertFalse(desc.getResult(String.class).isPresent()); + + RuntimeException failure = desc.getOutcomeFailure(); + assertNotNull(failure); + assertTrue(failure instanceof ApplicationFailure); + assertEquals("boom", ((ApplicationFailure) failure).getOriginalMessage()); + } + @Test public void testGetPriorityPresent() { io.temporal.api.common.v1.Priority protoPriority = io.temporal.api.common.v1.Priority.newBuilder().setPriorityKey(3).build(); ActivityExecutionInfo info = buildInfo("id", "run").toBuilder().setPriority(protoPriority).build(); - ActivityExecutionDescription desc = - new ActivityExecutionDescription(info, CONVERTER, "test-ns"); + ActivityExecutionDescription desc = describe(info); Priority priority = desc.getPriority(); assertNotNull(priority); diff --git a/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityOperatorCommandsTest.java b/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityOperatorCommandsTest.java new file mode 100644 index 0000000000..acf1f944da --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityOperatorCommandsTest.java @@ -0,0 +1,791 @@ +package io.temporal.client.functional; + +import static io.temporal.testUtils.Eventually.assertEventually; +import static org.junit.Assert.*; +import static org.junit.Assume.assumeTrue; + +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityExecutionContext; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.api.enums.v1.ActivityExecutionStatus; +import io.temporal.api.enums.v1.PendingActivityState; +import io.temporal.client.ActivityCanceledException; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.ActivityExecutionDescription; +import io.temporal.client.ActivityHandle; +import io.temporal.client.DescribeActivityOptions; +import io.temporal.client.PauseActivityOptions; +import io.temporal.client.ResetActivityOptions; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.UpdateActivityOptions; +import io.temporal.common.CancellationToken; +import io.temporal.common.Priority; +import io.temporal.common.RetryOptions; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor; +import io.temporal.common.interceptors.ActivityClientCallsInterceptor.*; +import io.temporal.common.interceptors.ActivityClientCallsInterceptorBase; +import io.temporal.common.interceptors.ActivityClientInterceptorBase; +import io.temporal.failure.ApplicationFailure; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import org.junit.Rule; +import org.junit.Test; + +/** + * Integration tests for the standalone-activity operator commands on {@link ActivityHandle}: pause, + * unpause, reset and updateOptions. Each asserts an observable server state change. + * + *

Gated behind {@link SDKTestWorkflowRule#useExternalService} because the embedded test server + * does not support the standalone activity APIs. + */ +public class StandaloneActivityOperatorCommandsTest { + + /** Heartbeat details are opt-in on describe; these tests assert on them. */ + private static final DescribeActivityOptions WITH_HEARTBEAT_DETAILS = + DescribeActivityOptions.newBuilder().setIncludeHeartbeatDetails(true).build(); + + // --------------------------------------------------------------------------- + // Activities + // --------------------------------------------------------------------------- + + /** Long-running activity that heartbeats and runs until cancellation/interruption. */ + @ActivityInterface + public interface SlowActivity { + @ActivityMethod(name = "Slow") + void run(); + } + + public static class SlowActivityImpl implements SlowActivity { + @Override + public void run() { + Activity.getExecutionContext().heartbeat(null); + while (true) { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + Activity.getExecutionContext().heartbeat(null); + } + } + } + + /** Takes two arguments, so a describe can read a multi-argument input back off the server. */ + @ActivityInterface + public interface TwoArgActivity { + @ActivityMethod(name = "TwoArg") + String run(String word, Integer count); + } + + public static class TwoArgActivityImpl implements TwoArgActivity { + @Override + public String run(String word, Integer count) { + return word + "-" + count; + } + } + + /** Returns immediately. Used with a start delay so it can be paused while scheduled. */ + @ActivityInterface + public interface QuickActivity { + @ActivityMethod(name = "Quick") + String run(); + } + + public static class QuickActivityImpl implements QuickActivity { + @Override + public String run() { + return "resumed"; + } + } + + /** Fails until the third attempt, then succeeds. Drives an activity past its first attempt. */ + @ActivityInterface + public interface FailThenSucceedActivity { + @ActivityMethod(name = "FailThenSucceed") + String run(); + } + + public static class FailThenSucceedActivityImpl implements FailThenSucceedActivity { + @Override + public String run() { + if (Activity.getExecutionContext().getInfo().getAttempt() < 3) { + throw ApplicationFailure.newFailure("retryable failure", "retry-type"); + } + return "done"; + } + } + + /** + * Records heartbeat details on attempt 1, then blocks waiting for cancellation. The heartbeat + * runs on its own — not adjacent to any completion RPC — so the details reliably persist and are + * observable via describe. Later attempts (after a reset or an unpause that spawns a new attempt) + * do not heartbeat, so any operator-driven clearing of the details stays observable. + */ + @ActivityInterface + public interface HeartbeatOnceActivity { + @ActivityMethod(name = "HeartbeatOnce") + void run(); + } + + public static class HeartbeatOnceActivityImpl implements HeartbeatOnceActivity { + @Override + public void run() { + ActivityExecutionContext ctx = Activity.getExecutionContext(); + if (ctx.getInfo().getAttempt() == 1) { + ctx.heartbeat("hb-details"); + } + CancellationToken token = ctx.getCancellationToken(); + while (!token.isCancellationRequested()) { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + } + } + + // --------------------------------------------------------------------------- + // Rule + helpers + // --------------------------------------------------------------------------- + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setActivityImplementations( + new SlowActivityImpl(), + new QuickActivityImpl(), + new FailThenSucceedActivityImpl(), + new TwoArgActivityImpl(), + new HeartbeatOnceActivityImpl()) + .build(); + + /** + * A running activity does not transition straight to PAUSED on pause: the server records + * PAUSE_REQUESTED and only moves to PAUSED once the worker drops the attempt. A long-running + * heartbeating activity that has not yet noticed the pause stays in PAUSE_REQUESTED, so both + * states count as "paused" for an observability assertion. + */ + private static final List PAUSED_STATES = + Arrays.asList( + PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED, + PendingActivityState.PENDING_ACTIVITY_STATE_PAUSE_REQUESTED); + + private String uniqueId() { + return "act-" + UUID.randomUUID(); + } + + private ActivityClient newActivityClient() { + return ActivityClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder().setNamespace(SDKTestWorkflowRule.NAMESPACE).build()); + } + + private void assertEventuallyPaused(ActivityHandle handle) { + assertEventually( + Duration.ofSeconds(30), + () -> + assertTrue( + "expected paused run state, got " + handle.describe().getRunState(), + PAUSED_STATES.contains(handle.describe().getRunState()))); + } + + /** Start a SlowActivity and wait until it has actually started running on the worker. */ + private ActivityHandle startRunningSlowActivity(StartActivityOptions.Builder optsBuilder) { + ActivityHandle handle = + newActivityClient().start(SlowActivity.class, SlowActivity::run, optsBuilder.build()); + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + PendingActivityState.PENDING_ACTIVITY_STATE_STARTED, + handle.describe().getRunState())); + return handle; + } + + /** + * Start a HeartbeatOnceActivity and wait until its first attempt has recorded heartbeat details. + * The activity keeps running (sleeping until interrupted) once heartbeat has fired, so pause + * transitions the activity through PAUSE_REQUESTED to PAUSED — assertEventuallyPaused tolerates + * both. + */ + private ActivityHandle startHeartbeatReadyActivity() { + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setHeartbeatTimeout(Duration.ofSeconds(30)) + .build(); + ActivityHandle handle = + newActivityClient().start(HeartbeatOnceActivity.class, HeartbeatOnceActivity::run, opts); + assertEventually( + Duration.ofSeconds(30), + () -> + assertTrue( + "expected heartbeat details to be recorded", + handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails())); + return handle; + } + + private StartActivityOptions.Builder slowOpts() { + return StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setHeartbeatTimeout(Duration.ofSeconds(30)); + } + + // --------------------------------------------------------------------------- + // Tests + // --------------------------------------------------------------------------- + + // Overrides the rule's default 10s global timeout: the start delay makes this take longer. + @Test(timeout = 60_000) + public void unpauseResumes() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityClient client = newActivityClient(); + // Start with a long delay so the activity sits SCHEDULED and can be paused before it runs. + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setStartDelay(Duration.ofSeconds(30)) + .build(); + ActivityHandle handle = client.start(QuickActivity.class, QuickActivity::run, opts); + + handle.pause(PauseActivityOptions.newBuilder().setReason("pause-before-unpause").build()); + // A not-yet-started (scheduled) activity transitions fully to PAUSED. + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED, + handle.describe().getRunState())); + + handle.unpause(); + // After unpause the activity proceeds and completes successfully (proving it resumed). + assertEquals("resumed", handle.getResult()); + } + + // Overrides the rule's default 10s global timeout: driving retries + reset takes longer. + @Test(timeout = 60_000) + public void reset() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityClient client = newActivityClient(); + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setRetryOptions( + RetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(200)) + .setBackoffCoefficient(1.0) + .setMaximumInterval(Duration.ofMillis(200)) + .setMaximumAttempts(50) + .build()) + .build(); + ActivityHandle handle = + client.start(FailThenSucceedActivity.class, FailThenSucceedActivity::run, opts); + + // Wait until the activity has recorded more than one attempt (i.e. it has retried). + assertEventually( + Duration.ofSeconds(30), + () -> assertTrue("expected attempt > 1 before reset", handle.describe().getAttempt() > 1)); + + handle.reset(); + + // After reset the attempt counter goes back to the start. + assertEventually( + Duration.ofSeconds(30), + () -> assertEquals("attempt should be reset to 1", 1, handle.describe().getAttempt())); + handle.terminate("cleanup"); + } + + @Test + public void updateOptionsRespectsMask() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = + startRunningSlowActivity( + slowOpts() + .setStartToCloseTimeout(Duration.ofSeconds(45)) + .setScheduleToCloseTimeout(Duration.ofSeconds(120))); + + UpdateActivityOptions updated = + handle.updateOptions( + UpdateActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(90)) + .build()); + + // Returned options: only start_to_close changed; schedule_to_close kept its original value. + assertEquals(Duration.ofSeconds(90), updated.getStartToCloseTimeout()); + assertEquals(Duration.ofSeconds(120), updated.getScheduleToCloseTimeout()); + + // Confirm via describe that the partial update was applied server-side. + assertEventually( + Duration.ofSeconds(30), + () -> { + ActivityExecutionDescription desc = handle.describe(); + assertEquals(Duration.ofSeconds(90), desc.getStartToCloseTimeout()); + assertEquals(Duration.ofSeconds(120), desc.getScheduleToCloseTimeout()); + }); + handle.terminate("cleanup"); + } + + // Overrides the rule's default 10s global timeout: uses a start delay to keep the activity + // scheduled while every option is updated and observed. + @Test(timeout = 60_000) + public void updateOptionsAllFields() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + // Start delayed so the activity stays SCHEDULED (never runs) while we update every option. + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setScheduleToCloseTimeout(Duration.ofSeconds(100)) + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .setStartDelay(Duration.ofSeconds(300)) + .build(); + ActivityHandle handle = + newActivityClient().start(QuickActivity.class, QuickActivity::run, opts); + + UpdateActivityOptions updated = + handle.updateOptions( + UpdateActivityOptions.newBuilder() + .setTaskQueue("updated-tq") + .setScheduleToCloseTimeout(Duration.ofSeconds(200)) + .setScheduleToStartTimeout(Duration.ofSeconds(15)) + .setStartToCloseTimeout(Duration.ofSeconds(90)) + .setHeartbeatTimeout(Duration.ofSeconds(25)) + .setRetryOptions( + RetryOptions.newBuilder() + .setInitialInterval(Duration.ofSeconds(1)) + .setBackoffCoefficient(2.0) + .setMaximumAttempts(7) + .build()) + .setPriority(Priority.newBuilder().setPriorityKey(3).build()) + .setStartDelay(Duration.ofSeconds(500)) + .build()); + + // Every field is settable and lands: the returned options reflect each new value. + assertEquals("updated-tq", updated.getTaskQueue()); + assertEquals(Duration.ofSeconds(200), updated.getScheduleToCloseTimeout()); + assertEquals(Duration.ofSeconds(15), updated.getScheduleToStartTimeout()); + assertEquals(Duration.ofSeconds(90), updated.getStartToCloseTimeout()); + assertEquals(Duration.ofSeconds(25), updated.getHeartbeatTimeout()); + assertEquals(7, updated.getRetryOptions().getMaximumAttempts()); + assertEquals(3, updated.getPriority().getPriorityKey()); + assertEquals(Duration.ofSeconds(500), updated.getStartDelay()); + + // And describe reflects them server-side. + ActivityExecutionDescription desc = handle.describe(); + assertEquals("updated-tq", desc.getTaskQueue()); + assertEquals(Duration.ofSeconds(200), desc.getScheduleToCloseTimeout()); + assertEquals(Duration.ofSeconds(15), desc.getScheduleToStartTimeout()); + assertEquals(Duration.ofSeconds(90), desc.getStartToCloseTimeout()); + assertEquals(Duration.ofSeconds(25), desc.getHeartbeatTimeout()); + assertEquals(7, desc.getRetryOptions().getMaximumAttempts()); + assertEquals(3, desc.getPriority().getPriorityKey()); + assertEquals(Duration.ofSeconds(500), desc.getStartDelay()); + // execution_time (api#807 + temporal#11017): reflects the updated start_delay. Server + // recomputes it on UpdateActivityOptions, so it lands at schedule_time + 500s (the new value), + // not schedule_time + 300s (the value at start). + assertEquals( + desc.getScheduledTime().plus(Duration.ofSeconds(500)).getEpochSecond(), + desc.getExecutionTime().getEpochSecond()); + + handle.terminate("cleanup"); + } + + @Test + public void updateOptionsRestoreOriginal() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = + startRunningSlowActivity(slowOpts().setStartToCloseTimeout(Duration.ofSeconds(45))); + + // Change an option away from the original. + UpdateActivityOptions changed = + handle.updateOptions( + UpdateActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(90)) + .build()); + assertEquals(Duration.ofSeconds(90), changed.getStartToCloseTimeout()); + + // restore_original alone reverts to the value the activity was created with. + UpdateActivityOptions restored = handle.restoreOriginalOptions(); + assertEquals(Duration.ofSeconds(45), restored.getStartToCloseTimeout()); + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void updateOptionsOnPausedActivity() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + // Start delayed so the activity sits SCHEDULED and pauses to a true PAUSED state rather than + // the PAUSE_REQUESTED a running activity lands in. + ActivityHandle handle = + newActivityClient() + .start( + QuickActivity.class, + QuickActivity::run, + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(45)) + .setScheduleToCloseTimeout(Duration.ofSeconds(120)) + .setStartDelay(Duration.ofSeconds(60)) + .build()); + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED, + handle.describe().getRunState())); + + // Updating options is legal while paused, and the new value lands. + UpdateActivityOptions updated = + handle.updateOptions( + UpdateActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(90)) + .build()); + assertEquals(Duration.ofSeconds(90), updated.getStartToCloseTimeout()); + + ActivityExecutionDescription desc = handle.describe(); + assertEquals(Duration.ofSeconds(90), desc.getStartToCloseTimeout()); + // The mask is still honored while paused — an option we didn't touch keeps its original value. + assertEquals(Duration.ofSeconds(120), desc.getScheduleToCloseTimeout()); + // And the update leaves the activity paused; it is not an implicit unpause. + assertEquals(PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED, desc.getRunState()); + assertEquals(ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_PAUSED, desc.getStatus()); + + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void resetKeepsPaused() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + // Start delayed so the activity sits SCHEDULED and pauses to a true PAUSED state (not the + // PAUSE_REQUESTED of a running activity), which is what keep_paused must preserve across reset. + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setStartDelay(Duration.ofSeconds(30)) + .build(); + ActivityHandle handle = + newActivityClient().start(QuickActivity.class, QuickActivity::run, opts); + + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED, + handle.describe().getRunState())); + + handle.reset(ResetActivityOptions.newBuilder().setKeepPaused(true).build()); + + // keep_paused keeps the activity paused across the reset. + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + "expected activity to stay paused after reset", + PendingActivityState.PENDING_ACTIVITY_STATE_PAUSED, + handle.describe().getRunState())); + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void resetRestoresOriginalOptions() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = + startRunningSlowActivity(slowOpts().setStartToCloseTimeout(Duration.ofSeconds(45))); + + UpdateActivityOptions updated = + handle.updateOptions( + UpdateActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(90)) + .build()); + assertEquals(Duration.ofSeconds(90), updated.getStartToCloseTimeout()); + + handle.reset(ResetActivityOptions.newBuilder().setRestoreOriginalOptions(true).build()); + + // restore_original_options reverts start_to_close back to the value the activity started with. + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + "start_to_close should be restored to original", + Duration.ofSeconds(45), + handle.describe().getStartToCloseTimeout())); + handle.terminate("cleanup"); + } + + /** + * The payload-bearing describe fields are opt-in (api#792). Assert the default really is "off" + * rather than the SDK quietly requesting everything: same activity, same moment, two describes. + */ + @Test(timeout = 60_000) + public void describePayloadFieldsAreOptIn() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = startHeartbeatReadyActivity(); + + assertFalse(handle.describe().hasHeartbeatDetails()); + assertEquals(0, handle.describe().getHeartbeatDetails().getSize()); + assertTrue(handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails()); + assertEquals( + "hb-details", + handle.describe(WITH_HEARTBEAT_DETAILS).getHeartbeatDetails().get(0, String.class)); + handle.terminate("cleanup"); + } + + /** + * Input and outcome are opt-in like the other payload fields. Uses a two-argument activity so + * {@link ActivityExecutionDescription#getInput(int, Class)} has more than one argument to read. + */ + @Test(timeout = 60_000) + public void describeReadsInputAndOutcome() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .build(); + ActivityHandle handle = + newActivityClient().start(TwoArgActivity.class, TwoArgActivity::run, opts, "ping", 7); + assertEquals("ping-7", handle.getResult(String.class)); + + // Default describe omits both. + ActivityExecutionDescription bare = handle.describe(); + assertFalse(bare.hasInput()); + assertEquals(0, bare.getInput().getSize()); + assertFalse(bare.hasResult()); + assertNull(bare.getOutcomeFailure()); + + ActivityExecutionDescription desc = + handle.describe( + DescribeActivityOptions.newBuilder() + .setIncludeInput(true) + .setIncludeOutcome(true) + .build()); + assertTrue(desc.hasInput()); + assertEquals(2, desc.getInput().getSize()); + assertEquals("ping", desc.getInput().get(0, String.class)); + assertEquals(Integer.valueOf(7), desc.getInput().get(1, Integer.class)); + assertTrue(desc.hasResult()); + assertEquals("ping-7", desc.getResult(String.class).orElse(null)); + // A successful outcome has no failure arm. + assertNull(desc.getOutcomeFailure()); + } + + /** The other arm of the outcome oneof: a terminally failed activity has a failure, no result. */ + @Test(timeout = 60_000) + public void describeReadsFailureOutcome() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + StartActivityOptions opts = + StartActivityOptions.newBuilder() + .setId(uniqueId()) + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(1).build()) + .build(); + ActivityHandle handle = + newActivityClient() + .start(FailThenSucceedActivity.class, FailThenSucceedActivity::run, opts); + assertThrows(Exception.class, () -> handle.getResult(String.class)); + + ActivityExecutionDescription desc = + handle.describe(DescribeActivityOptions.newBuilder().setIncludeOutcome(true).build()); + assertFalse(desc.hasResult()); + assertFalse(desc.getResult(String.class).isPresent()); + + RuntimeException failure = desc.getOutcomeFailure(); + assertNotNull(failure); + assertTrue(failure instanceof ApplicationFailure); + assertEquals("retryable failure", ((ApplicationFailure) failure).getOriginalMessage()); + } + + @Test(timeout = 60_000) + public void pausePreservesHeartbeat() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = startHeartbeatReadyActivity(); + + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventuallyPaused(handle); + + // Pause never touches heartbeat details — they persist across the transition. + assertTrue( + "heartbeat details should be preserved across pause", + handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails()); + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void unpausePreservesHeartbeat() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = startHeartbeatReadyActivity(); + + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventuallyPaused(handle); + + // Unpause preserves heartbeat details. The re-dispatched attempt doesn't heartbeat (only + // attempt 1 does), so the persisted details are stable and observable. + handle.unpause(); + + assertEventually( + Duration.ofSeconds(30), + () -> + assertTrue( + "heartbeat details should be preserved after unpause", + handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails())); + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void resetPreservesHeartbeatByDefault() throws InterruptedException { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = startHeartbeatReadyActivity(); + + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventuallyPaused(handle); + + // As of api#848 / temporal#11417, reset does NOT clear heartbeat details by default — + // you must pass resetHeartbeat=true. keep_paused so no new attempt reshapes state. + handle.reset(ResetActivityOptions.newBuilder().setKeepPaused(true).build()); + // Give the server time to persist any state change, then confirm details survive. + Thread.sleep(2000); + assertTrue( + "heartbeat details should be preserved after default reset", + handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails()); + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void resetClearsHeartbeatWhenFlagSet() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = startHeartbeatReadyActivity(); + + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventuallyPaused(handle); + + // Opt-in flag clears details. + handle.reset( + ResetActivityOptions.newBuilder().setKeepPaused(true).setResetHeartbeat(true).build()); + + assertEventually( + Duration.ofSeconds(30), + () -> + assertFalse( + "heartbeat details should be cleared after reset(reset_heartbeat)", + handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails())); + handle.terminate("cleanup"); + } + + @Test(timeout = 60_000) + public void updateOptionsPreservesHeartbeat() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + ActivityHandle handle = startHeartbeatReadyActivity(); + + handle.pause(PauseActivityOptions.newBuilder().setReason("hold").build()); + assertEventuallyPaused(handle); + + // UpdateOptions changes activity options only; it never touches heartbeat details. + handle.updateOptions( + UpdateActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(90)).build()); + + assertTrue( + "heartbeat details should be preserved after updateOptions", + handle.describe(WITH_HEARTBEAT_DETAILS).hasHeartbeatDetails()); + handle.terminate("cleanup"); + } + + // Overrides the rule's default 10s global timeout: exercises every command against a real server. + @Test(timeout = 60_000) + public void interceptorInvokesEachOperatorCommand() { + assumeTrue(SDKTestWorkflowRule.useExternalService); + List events = Collections.synchronizedList(new ArrayList<>()); + ActivityClient client = + ActivityClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder() + .setNamespace(SDKTestWorkflowRule.NAMESPACE) + .setInterceptors(Collections.singletonList(new RecordingInterceptor(events))) + .build()); + + ActivityHandle handle = + client.start(SlowActivity.class, SlowActivity::run, slowOpts().build()); + assertEventually( + Duration.ofSeconds(30), + () -> + assertEquals( + PendingActivityState.PENDING_ACTIVITY_STATE_STARTED, + handle.describe().getRunState())); + + handle.pause(PauseActivityOptions.newBuilder().setReason("reason").build()); + assertEventuallyPaused(handle); + handle.unpause(); + handle.updateOptions( + UpdateActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(90)).build()); + handle.reset(); + handle.terminate("cleanup"); + + assertTrue("pause should flow through the interceptor", events.contains("pause")); + assertTrue("unpause should flow through the interceptor", events.contains("unpause")); + assertTrue("reset should flow through the interceptor", events.contains("reset")); + assertTrue( + "updateOptions should flow through the interceptor", events.contains("updateOptions")); + } + + /** Records each operator command as it flows through the client interceptor chain. */ + private static class RecordingInterceptor extends ActivityClientInterceptorBase { + private final List events; + + RecordingInterceptor(List events) { + this.events = events; + } + + @Override + public ActivityClientCallsInterceptor activityClientCallsInterceptor( + ActivityClientCallsInterceptor next) { + return new ActivityClientCallsInterceptorBase(next) { + @Override + public PauseActivityOutput pauseActivity(PauseActivityInput input) { + events.add("pause"); + return super.pauseActivity(input); + } + + @Override + public UnpauseActivityOutput unpauseActivity(UnpauseActivityInput input) { + events.add("unpause"); + return super.unpauseActivity(input); + } + + @Override + public ResetActivityOutput resetActivity(ResetActivityInput input) { + events.add("reset"); + return super.resetActivity(input); + } + + @Override + public UpdateActivityOptionsOutput updateActivityOptions(UpdateActivityOptionsInput input) { + events.add("updateOptions"); + return super.updateActivityOptions(input); + } + }; + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityTest.java b/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityTest.java index 5be3226dcd..2a7c87c694 100644 --- a/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityTest.java +++ b/temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityTest.java @@ -854,7 +854,9 @@ public void testDescribeLastFailureIsPopulatedDuringRetryBackoff() { assertEventually( Duration.ofSeconds(60), () -> { - ActivityExecutionDescription desc = handle.describe(); + ActivityExecutionDescription desc = + handle.describe( + DescribeActivityOptions.newBuilder().setIncludeLastFailure(true).build()); Exception lastFailure = desc.getLastFailure(); assertNotNull("last_failure should be set after a failed attempt", lastFailure); assertThat(lastFailure, instanceOf(ApplicationFailure.class)); diff --git a/temporal-sdk/src/test/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBaseTest.java b/temporal-sdk/src/test/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBaseTest.java index e3cc99b3a1..400bb2ce9a 100644 --- a/temporal-sdk/src/test/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBaseTest.java +++ b/temporal-sdk/src/test/java/io/temporal/common/interceptors/ActivityClientCallsInterceptorBaseTest.java @@ -6,6 +6,7 @@ import io.temporal.client.ActivityExecutionCount; import io.temporal.client.ActivityExecutionDescription; import io.temporal.client.ActivityExecutionMetadata; +import io.temporal.client.DescribeActivityOptions; import io.temporal.client.StartActivityOptions; import io.temporal.common.interceptors.ActivityClientCallsInterceptor.*; import java.time.Duration; @@ -89,7 +90,8 @@ public void testDescribeActivityDelegatesToNext() { DescribeActivityOutput output = new DescribeActivityOutput(desc); when(next.describeActivity(any(DescribeActivityInput.class))).thenReturn(output); - DescribeActivityInput input = new DescribeActivityInput("id", null); + DescribeActivityInput input = + new DescribeActivityInput("id", null, DescribeActivityOptions.getDefaultInstance()); DescribeActivityOutput result = base.describeActivity(input); assertSame(output, result); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java new file mode 100644 index 0000000000..298742392a --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/ActivityHandleOperatorCommandsTest.java @@ -0,0 +1,124 @@ +package io.temporal.internal.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.temporal.api.workflowservice.v1.PauseActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.ResetActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.UnpauseActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.UpdateActivityExecutionOptionsRequest; +import io.temporal.api.workflowservice.v1.UpdateActivityExecutionOptionsResponse; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.PauseActivityOptions; +import io.temporal.client.ResetActivityOptions; +import io.temporal.client.UnpauseActivityOptions; +import io.temporal.client.UntypedActivityHandle; +import io.temporal.client.UpdateActivityOptions; +import io.temporal.internal.client.external.GenericWorkflowClient; +import java.time.Duration; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** Unit test for the operator-command request fields that the server does not surface back. */ +public class ActivityHandleOperatorCommandsTest { + + private final GenericWorkflowClient genericClient = mock(GenericWorkflowClient.class); + + private final ActivityClientOptions clientOptions = + ActivityClientOptions.newBuilder() + .setNamespace("test-namespace") + .setIdentity("test-identity") + .build(); + + private UntypedActivityHandle newHandle() { + return new ActivityHandleImpl( + "act-1", "run-1", new RootActivityClientInvoker(genericClient, clientOptions)); + } + + @Test + public void unobservableRequestFields() { + // updateActivityOptions returns a non-void response; stub so handle.updateOptions doesn't NPE. + when(genericClient.updateActivityOptions(any())) + .thenReturn(UpdateActivityExecutionOptionsResponse.getDefaultInstance()); + + UntypedActivityHandle handle = newHandle(); + + handle.pause(PauseActivityOptions.newBuilder().setReason("because").build()); + handle.unpause( + UnpauseActivityOptions.newBuilder() + .setReason("go") + .setJitter(Duration.ofSeconds(5)) + .build()); + handle.reset( + ResetActivityOptions.newBuilder() + .setJitter(Duration.ofSeconds(2)) + .setResetHeartbeat(true) + .build()); + handle.updateOptions( + UpdateActivityOptions.newBuilder().setStartDelay(Duration.ofSeconds(7)).build()); + + // pause carries the reason and an auto-generated dedup request_id; neither is returned by + // describe. + PauseActivityExecutionRequest pauseReq = capturePause(); + assertEquals("because", pauseReq.getReason()); + assertTrue("pause request_id should be set", !pauseReq.getRequestId().isEmpty()); + + // unpause carries the reason, jitter, and an auto-generated dedup request_id (api#844). + UnpauseActivityExecutionRequest unpauseReq = captureUnpause(); + assertEquals("go", unpauseReq.getReason()); + assertEquals(5, unpauseReq.getJitter().getSeconds()); + assertEquals(0, unpauseReq.getJitter().getNanos()); + assertTrue("unpause request_id should be set", !unpauseReq.getRequestId().isEmpty()); + + // reset carries jitter, an auto-generated dedup request_id (api#844), and reset_heartbeat + // (api#848). + ResetActivityExecutionRequest resetReq = captureReset(); + assertEquals(2, resetReq.getJitter().getSeconds()); + assertEquals(0, resetReq.getJitter().getNanos()); + assertTrue("reset request_id should be set", !resetReq.getRequestId().isEmpty()); + assertTrue("reset should carry reset_heartbeat=true", resetReq.getResetHeartbeat()); + + // updateOptions carries start_delay in activity_options with a matching update_mask path, plus + // an auto-generated dedup request_id (api#844). start_delay is applied server-side but not + // otherwise observable from the request. + UpdateActivityExecutionOptionsRequest updateReq = captureUpdate(); + assertEquals(7, updateReq.getActivityOptions().getStartDelay().getSeconds()); + assertEquals(0, updateReq.getActivityOptions().getStartDelay().getNanos()); + assertTrue( + "update_mask should include start_delay", + updateReq.getUpdateMask().getPathsList().contains("start_delay")); + assertTrue("updateOptions request_id should be set", !updateReq.getRequestId().isEmpty()); + } + + private PauseActivityExecutionRequest capturePause() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(PauseActivityExecutionRequest.class); + verify(genericClient).pauseActivity(captor.capture()); + return captor.getValue(); + } + + private UnpauseActivityExecutionRequest captureUnpause() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(UnpauseActivityExecutionRequest.class); + verify(genericClient).unpauseActivity(captor.capture()); + return captor.getValue(); + } + + private ResetActivityExecutionRequest captureReset() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(ResetActivityExecutionRequest.class); + verify(genericClient).resetActivity(captor.capture()); + return captor.getValue(); + } + + private UpdateActivityExecutionOptionsRequest captureUpdate() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(UpdateActivityExecutionOptionsRequest.class); + verify(genericClient).updateActivityOptions(captor.capture()); + return captor.getValue(); + } +}