diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java index c92d1d8390..42dc067497 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java @@ -110,8 +110,9 @@ public static WorkflowClient newInstance( .getMetricsScope() .tagged(MetricsTag.defaultTags(options.getNamespace())); ExternalStorage externalStorage = options.getExternalStorage(); - this.externalStorageRunner = + ExternalStorageRunner externalStorageRunner = externalStorage == null ? null : ExternalStorageRunner.create(externalStorage); + this.externalStorageRunner = externalStorageRunner; this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope); this.interceptors = options.getInterceptors(); this.workflowClientCallsInvoker = initializeClientInvoker(); @@ -120,7 +121,8 @@ public static WorkflowClient newInstance( workflowServiceStubs, options.getNamespace(), options.getIdentity(), - options.getDataConverter()); + options.getDataConverter(), + externalStorageRunner); java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval(); if (!heartbeatInterval.isNegative()) { @@ -133,7 +135,8 @@ public static WorkflowClient newInstance( private WorkflowClientCallsInterceptor initializeClientInvoker() { WorkflowClientCallsInterceptor workflowClientInvoker = - new RootWorkflowClientInvoker(genericClient, options, workerFactoryRegistry); + new RootWorkflowClientInvoker( + genericClient, options, workerFactoryRegistry, externalStorageRunner); for (WorkflowClientInterceptor clientInterceptor : interceptors) { workflowClientInvoker = clientInterceptor.workflowClientCallsInterceptor(workflowClientInvoker); diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionDescription.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionDescription.java index 6f54031a97..37122381c7 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionDescription.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionDescription.java @@ -1,5 +1,6 @@ package io.temporal.client; +import io.temporal.api.common.v1.Payload; import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionResponse; import io.temporal.common.converter.DataConverter; import io.temporal.payload.context.WorkflowSerializationContext; @@ -29,15 +30,13 @@ public String getStaticSummary() { if (!response.getExecutionConfig().getUserMetadata().hasSummary()) { return null; } + Payload summary = response.getExecutionConfig().getUserMetadata().getSummary(); return dataConverter .withContext( new WorkflowSerializationContext( response.getWorkflowExecutionInfo().getParentNamespaceId(), response.getWorkflowExecutionInfo().getExecution().getWorkflowId())) - .fromPayload( - response.getExecutionConfig().getUserMetadata().getSummary(), - String.class, - String.class); + .fromPayload(summary, String.class, String.class); } /** @@ -51,15 +50,13 @@ public String getStaticDetails() { if (!response.getExecutionConfig().getUserMetadata().hasDetails()) { return null; } + Payload details = response.getExecutionConfig().getUserMetadata().getDetails(); return dataConverter .withContext( new WorkflowSerializationContext( response.getWorkflowExecutionInfo().getParentNamespaceId(), response.getWorkflowExecutionInfo().getExecution().getWorkflowId())) - .fromPayload( - response.getExecutionConfig().getUserMetadata().getDetails(), - String.class, - String.class); + .fromPayload(details, String.class, String.class); } /** Returns the raw response from the Temporal service. */ diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java index 4acc1d17dd..5f9584180f 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextFactoryImpl.java @@ -4,6 +4,7 @@ import io.temporal.client.WorkflowClient; import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import java.nio.ByteBuffer; import java.time.Duration; import java.util.Arrays; @@ -11,6 +12,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; +import javax.annotation.Nullable; public class ActivityExecutionContextFactoryImpl implements ActivityExecutionContextFactory { private final WorkflowClient client; @@ -21,6 +23,7 @@ public class ActivityExecutionContextFactoryImpl implements ActivityExecutionCon private final DataConverter dataConverter; private final ScheduledExecutorService heartbeatExecutor; private final ManualActivityCompletionClientFactory manualCompletionClientFactory; + private final @Nullable ExternalStorageRunner externalStorage; private final ConcurrentMap activeContexts = new ConcurrentHashMap<>(); @@ -31,7 +34,8 @@ public ActivityExecutionContextFactoryImpl( Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, DataConverter dataConverter, - ScheduledExecutorService heartbeatExecutor) { + ScheduledExecutorService heartbeatExecutor, + @Nullable ExternalStorageRunner externalStorage) { this.client = Objects.requireNonNull(client); this.identity = identity; this.namespace = Objects.requireNonNull(namespace); @@ -40,9 +44,10 @@ public ActivityExecutionContextFactoryImpl( Objects.requireNonNull(defaultHeartbeatThrottleInterval); this.dataConverter = Objects.requireNonNull(dataConverter); this.heartbeatExecutor = Objects.requireNonNull(heartbeatExecutor); + this.externalStorage = externalStorage; this.manualCompletionClientFactory = ManualActivityCompletionClientFactory.newFactory( - client.getWorkflowServiceStubs(), namespace, identity, dataConverter); + client.getWorkflowServiceStubs(), namespace, identity, dataConverter, externalStorage); } @Override @@ -63,7 +68,8 @@ public InternalActivityExecutionContext createContext( identity, maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, - () -> cleanupContext(info.getTaskToken(), false)); + () -> cleanupContext(info.getTaskToken(), false), + externalStorage); activeContexts.put(taskToken, context); return context; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java index 40fe45c326..64cdd75383 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java @@ -10,6 +10,7 @@ import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.workflow.Functions; import java.lang.reflect.Type; @@ -18,6 +19,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import javax.annotation.Nullable; import javax.annotation.concurrent.ThreadSafe; /** @@ -55,7 +57,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext { String identity, Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, - Functions.Proc closeCallback) { + Functions.Proc closeCallback, + @Nullable ExternalStorageRunner externalStorage) { this.client = client; this.activity = activity; this.metricsScope = metricsScope; @@ -73,7 +76,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext { metricsScope, identity, maxHeartbeatThrottleInterval, - defaultHeartbeatThrottleInterval); + defaultHeartbeatThrottleInterval, + externalStorage); } /** @@ -155,7 +159,10 @@ public ManualActivityCompletionClient useLocalManualCompletion() { new ActivitySerializationContext(info); return new CompletionAwareManualCompletionClient( manualCompletionClientFactory.getClient( - info.getTaskToken(), metricsScope, activitySerializationContext), + info.getTaskToken(), + metricsScope, + activitySerializationContext, + HeartbeatContextImpl.storageTargetForActivity(info.getNamespace(), info)), completionHandle); } finally { lock.unlock(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java index 91da94ab0a..b7ec1dd10f 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java @@ -1,5 +1,7 @@ package io.temporal.internal.activity; +import com.google.common.base.Strings; +import com.google.protobuf.ByteString; import com.uber.m3.tally.Scope; import io.grpc.Status; import io.grpc.StatusRuntimeException; @@ -7,6 +9,7 @@ import io.temporal.activity.ActivityInfo; import io.temporal.api.common.v1.Payloads; import io.temporal.api.enums.v1.TimeoutType; +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; import io.temporal.client.*; import io.temporal.common.CancellationToken; @@ -14,16 +17,21 @@ import io.temporal.failure.TimeoutFailure; import io.temporal.internal.client.ActivityClientHelper; import io.temporal.internal.concurrent.structured.CancelSource; +import io.temporal.internal.payload.storage.ActivityStorageTargets; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import java.lang.reflect.Type; import java.time.Duration; import java.util.Optional; +import java.util.concurrent.CancellationException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import javax.annotation.Nullable; import javax.annotation.concurrent.ThreadSafe; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,6 +66,7 @@ static long getLocalHeartbeatTimeoutBufferMillis() { private final long heartbeatIntervalMillis; private final DataConverter dataConverter; private final DataConverter dataConverterWithActivityContext; + private final @Nullable ExternalStorageRunner externalStorage; private final Scope metricsScope; private final Optional prevAttemptHeartbeatDetails; @@ -89,7 +98,8 @@ public HeartbeatContextImpl( Scope metricsScope, String identity, Duration maxHeartbeatThrottleInterval, - Duration defaultHeartbeatThrottleInterval) { + Duration defaultHeartbeatThrottleInterval, + @Nullable ExternalStorageRunner externalStorage) { this( service, namespace, @@ -100,6 +110,7 @@ public HeartbeatContextImpl( identity, maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, + externalStorage, getLocalHeartbeatTimeoutBufferMillis()); } @@ -113,10 +124,12 @@ public HeartbeatContextImpl( String identity, Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, + @Nullable ExternalStorageRunner externalStorage, long localHeartbeatTimeoutBufferMillis) { this.service = service; this.metricsScope = metricsScope; this.dataConverter = dataConverter; + this.externalStorage = externalStorage; this.dataConverterWithActivityContext = dataConverter.withContext( new ActivitySerializationContext( @@ -330,16 +343,59 @@ private void checkHeartbeatTimeoutDeadlineLocked() { } } + private StorageDriverTargetInfo activityStorageTarget() { + return storageTargetForActivity(namespace, info); + } + + /** + * Standalone activities target the activity; workflow activities target their workflow, matching + * where {@link io.temporal.internal.worker.ActivityWorker} stores the activity task payloads. A + * non-empty {@code activityRunId} marks a standalone activity. + */ + static StorageDriverTargetInfo storageTargetForActivity(String namespace, ActivityInfo info) { + return ActivityStorageTargets.newBuilder(namespace) + .setActivity(info.getActivityId(), info.getActivityRunId(), info.getActivityType()) + .setWorkflow( + Strings.emptyToNull(info.getWorkflowId()), + Strings.emptyToNull(info.getWorkflowRunId()), + info.getWorkflowType()) + .build(); + } + + /** + * Offloads large heartbeat payloads aborting if the store call runs longer than the heartbeat + * interval or if the activity is cancelled. + */ + private void offloadHeartbeat(RecordActivityTaskHeartbeatRequest.Builder builder) { + CancelSource offloadCancel = + new CancelSource<>(CancellationException::new); + ScheduledFuture timeout = + heartbeatExecutor.schedule( + (Runnable) offloadCancel::cancel, heartbeatIntervalMillis, TimeUnit.MILLISECONDS); + CancellationToken.Registration onActivityCancel = + cancellationSource.token().onCancel(offloadCancel::cancel); + try { + externalStorage.store(builder, activityStorageTarget(), null, offloadCancel.token()); + } finally { + timeout.cancel(false); + onActivityCancel.close(); + } + } + private void sendHeartbeatRequest(Object details) { try { + RecordActivityTaskHeartbeatRequest.Builder builder = + RecordActivityTaskHeartbeatRequest.newBuilder() + .setTaskToken(ByteString.copyFrom(info.getTaskToken())) + .setNamespace(namespace) + .setIdentity(identity); + dataConverterWithActivityContext.toPayloads(details).ifPresent(builder::setDetails); + if (externalStorage != null) { + offloadHeartbeat(builder); + } + RecordActivityTaskHeartbeatRequest request = builder.build(); RecordActivityTaskHeartbeatResponse status = - ActivityClientHelper.sendHeartbeatRequest( - service, - namespace, - identity, - info.getTaskToken(), - dataConverterWithActivityContext.toPayloads(details), - metricsScope); + ActivityClientHelper.sendHeartbeatRequest(service, request, metricsScope); if (status.getCancelRequested()) { requestCancelLocked(); } else if (status.getActivityReset()) { diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java index eb3e98107c..cedbf38684 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java @@ -2,19 +2,13 @@ import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY; -import com.google.common.base.Preconditions; -import com.google.protobuf.ByteString; import com.uber.m3.tally.Scope; import io.temporal.activity.ManualActivityCompletionClient; -import io.temporal.api.common.v1.Payloads; -import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdResponse; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; import io.temporal.serviceclient.WorkflowServiceStubs; -import java.util.Optional; -import javax.annotation.Nonnull; /** * Contains methods that could but didn't become a part of the main {@link @@ -26,43 +20,21 @@ private ActivityClientHelper() {} public static RecordActivityTaskHeartbeatResponse sendHeartbeatRequest( WorkflowServiceStubs service, - String namespace, - String identity, - byte[] taskToken, - Optional payloads, + RecordActivityTaskHeartbeatRequest request, Scope metricsScope) { - RecordActivityTaskHeartbeatRequest.Builder request = - RecordActivityTaskHeartbeatRequest.newBuilder() - .setTaskToken(ByteString.copyFrom(taskToken)) - .setNamespace(namespace) - .setIdentity(identity); - payloads.ifPresent(request::setDetails); return service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .recordActivityTaskHeartbeat(request.build()); + .recordActivityTaskHeartbeat(request); } public static RecordActivityTaskHeartbeatByIdResponse recordActivityTaskHeartbeatById( WorkflowServiceStubs service, - String namespace, - String identity, - WorkflowExecution execution, - @Nonnull String activityId, - Optional payloads, + RecordActivityTaskHeartbeatByIdRequest request, Scope metricsScope) { - Preconditions.checkNotNull(activityId, "Either activity id or task token are required"); - RecordActivityTaskHeartbeatByIdRequest.Builder request = - RecordActivityTaskHeartbeatByIdRequest.newBuilder() - .setRunId(execution.getRunId()) - .setWorkflowId(execution.getWorkflowId()) - .setActivityId(activityId) - .setNamespace(namespace) - .setIdentity(identity); - payloads.ifPresent(request::setDetails); return service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .recordActivityTaskHeartbeatById(request.build()); + .recordActivityTaskHeartbeatById(request); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java index 502c12e8ee..22a45633d5 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootWorkflowClientInvoker.java @@ -10,6 +10,8 @@ import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.temporal.api.common.v1.*; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.enums.v1.QueryRejectCondition; import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage; import io.temporal.api.enums.v1.WorkflowExecutionStatus; import io.temporal.api.errordetails.v1.MultiOperationExecutionFailure; @@ -28,8 +30,8 @@ import io.temporal.internal.nexus.InternalNexusOperationContext; import io.temporal.internal.nexus.NexusOperationMetadata; import io.temporal.internal.nexus.OperationTokenUtil; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.worker.WorkerVersioningProtoUtils; -import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.serviceclient.StatusUtils; import io.temporal.worker.WorkflowTaskDispatchHandle; import java.lang.reflect.Type; @@ -47,28 +49,52 @@ public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor private static final long POLL_UPDATE_TIMEOUT_S = 60L; private final GenericWorkflowClient genericClient; - private final WorkflowClientOptions clientOptions; + private final String namespace; + private final String identity; + private final QueryRejectCondition queryRejectCondition; private final EagerWorkflowTaskDispatcher eagerWorkflowTaskDispatcher; private final WorkflowClientRequestFactory requestsHelper; + private final WorkflowClientDataConverterFactory converterFactory; public RootWorkflowClientInvoker( GenericWorkflowClient genericClient, WorkflowClientOptions clientOptions, WorkerFactoryRegistry workerFactoryRegistry) { + this(genericClient, clientOptions, workerFactoryRegistry, null); + } + + public RootWorkflowClientInvoker( + GenericWorkflowClient genericClient, + WorkflowClientOptions clientOptions, + WorkerFactoryRegistry workerFactoryRegistry, + @Nullable ExternalStorageRunner externalStorage) { + this.converterFactory = new WorkflowClientDataConverterFactory(clientOptions, externalStorage); this.genericClient = genericClient; - this.clientOptions = clientOptions; + this.namespace = clientOptions.getNamespace(); + this.identity = clientOptions.getIdentity(); + this.queryRejectCondition = clientOptions.getQueryRejectCondition(); this.eagerWorkflowTaskDispatcher = new EagerWorkflowTaskDispatcher(workerFactoryRegistry); this.requestsHelper = new WorkflowClientRequestFactory(clientOptions); } + private DataConverter workflowConverter(WorkflowExecution execution) { + return workflowConverter(execution, null); + } + + private DataConverter workflowConverter( + WorkflowExecution execution, @Nullable String workflowType) { + return workflowConverter(execution.getWorkflowId(), execution.getRunId(), workflowType); + } + + private DataConverter workflowConverter( + String workflowId, @Nullable String runId, @Nullable String workflowType) { + return converterFactory.forWorkflow(workflowId, runId, workflowType); + } + @Override public WorkflowStartOutput start(WorkflowStartInput input) { DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowId())); + workflowConverter(input.getWorkflowId(), null, input.getWorkflowType()); StartWorkflowExecutionRequest.Builder startRequest = toStartRequest(dataConverterWithWorkflowContext, input); @@ -126,8 +152,8 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) { SignalWorkflowExecutionRequest.newBuilder() .setSignalName(input.getSignalName()) .setWorkflowExecution(input.getWorkflowExecution()) - .setIdentity(clientOptions.getIdentity()) - .setNamespace(clientOptions.getNamespace()) + .setIdentity(identity) + .setNamespace(namespace) .setRequestId(UUID.randomUUID().toString()) .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)); @@ -138,12 +164,7 @@ public WorkflowSignalOutput signal(WorkflowSignalInput input) { request.addAllLinks(CurrentNexusOperationContext.get().getRequestLinks()); } - DataConverter dataConverterWitSignalContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + DataConverter dataConverterWitSignalContext = workflowConverter(input.getWorkflowExecution()); Optional inputArgs = dataConverterWitSignalContext.toPayloads(input.getArguments()); inputArgs.ifPresent(request::setInput); @@ -161,11 +182,8 @@ public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInpu WorkflowStartInput workflowStartInput = input.getWorkflowStartInput(); DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), workflowStartInput.getWorkflowId())); + workflowConverter( + workflowStartInput.getWorkflowId(), null, workflowStartInput.getWorkflowType()); StartWorkflowExecutionRequestOrBuilder startRequest = toStartRequest(dataConverterWithWorkflowContext, workflowStartInput); @@ -204,15 +222,11 @@ public WorkflowUpdateWithStartOutput updateWithStart( WorkflowStartInput startInput = input.getWorkflowStartInput(); DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), startInput.getWorkflowId())); + workflowConverter(startInput.getWorkflowId(), null, startInput.getWorkflowType()); ExecuteMultiOperationRequest request = ExecuteMultiOperationRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) + .setNamespace(namespace) .addOperations( 0, ExecuteMultiOperationRequest.Operation.newBuilder() @@ -355,11 +369,7 @@ private StartWorkflowExecutionRequest.Builder toStartRequest( @Override public GetResultOutput getResult(GetResultInput input) throws TimeoutException { DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter(input.getWorkflowExecution()); Optional resultValue = WorkflowClientLongPollHelper.getWorkflowExecutionResult( genericClient, @@ -380,11 +390,7 @@ public GetResultOutput getResult(GetResultInput input) throws TimeoutE @Override public GetResultAsyncOutput getResultAsync(GetResultInput input) { DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter(input.getWorkflowExecution()); CompletableFuture> resultValue = WorkflowClientLongPollAsyncHelper.getWorkflowExecutionResultAsync( genericClient, @@ -411,24 +417,20 @@ public QueryOutput query(QueryInput input) { .setQueryType(input.getQueryType()) .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)); DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter(input.getWorkflowExecution()); Optional inputArgs = dataConverterWithWorkflowContext.toPayloads(input.getArguments()); inputArgs.ifPresent(query::setQueryArgs); QueryWorkflowRequest request = QueryWorkflowRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) + .setNamespace(namespace) .setExecution( WorkflowExecution.newBuilder() .setWorkflowId(input.getWorkflowExecution().getWorkflowId()) .setRunId(input.getWorkflowExecution().getRunId())) .setQuery(query) - .setQueryRejectCondition(clientOptions.getQueryRejectCondition()) + .setQueryRejectCondition(queryRejectCondition) .build(); QueryWorkflowResponse result; @@ -451,11 +453,7 @@ public QueryOutput query(QueryInput input) { @Override public WorkflowUpdateHandle startUpdate(StartUpdateInput input) { DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter(input.getWorkflowExecution()); UpdateWorkflowExecutionRequest updateRequest = toUpdateWorkflowExecutionRequest(input, dataConverterWithWorkflowContext); @@ -514,10 +512,7 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( Request.Builder requestBuilder = Request.newBuilder() - .setMeta( - Meta.newBuilder() - .setUpdateId(input.getUpdateId()) - .setIdentity(clientOptions.getIdentity())) + .setMeta(Meta.newBuilder().setUpdateId(input.getUpdateId()).setIdentity(identity)) .setInput(updateInput); // If this update is being issued via TemporalNexusClientImpl.startWorkflowUpdate, @@ -530,7 +525,7 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( try { nexusOperationMetadata.operationToken = OperationTokenUtil.generateWorkflowUpdateOperationToken( - clientOptions.getNamespace(), + namespace, input.getWorkflowExecution().getWorkflowId(), input.getWorkflowExecution().getRunId(), input.getUpdateId()); @@ -555,7 +550,7 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( Request request = requestBuilder.build(); return UpdateWorkflowExecutionRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) + .setNamespace(namespace) .setWaitPolicy(input.getWaitPolicy()) .setWorkflowExecution( WorkflowExecution.newBuilder() @@ -622,11 +617,7 @@ private WorkflowUpdateHandle toUpdateHandle( @Override public PollWorkflowUpdateOutput pollWorkflowUpdate(PollWorkflowUpdateInput input) { DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter(input.getWorkflowExecution()); UpdateRef update = UpdateRef.newBuilder() @@ -643,8 +634,8 @@ public PollWorkflowUpdateOutput pollWorkflowUpdate(PollWorkflowUpdateInpu PollWorkflowExecutionUpdateRequest pollUpdateRequest = PollWorkflowExecutionUpdateRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) - .setIdentity(clientOptions.getIdentity()) + .setNamespace(namespace) + .setIdentity(identity) .setUpdateRef(update) .setWaitPolicy(waitPolicy) .build(); @@ -720,8 +711,8 @@ public CancelOutput cancel(CancelInput input) { RequestCancelWorkflowExecutionRequest.newBuilder() .setRequestId(UUID.randomUUID().toString()) .setWorkflowExecution(input.getWorkflowExecution()) - .setNamespace(clientOptions.getNamespace()) - .setIdentity(clientOptions.getIdentity()); + .setNamespace(namespace) + .setIdentity(identity); if (input.getReason() != null) { request.setReason(input.getReason()); } @@ -736,8 +727,8 @@ public CancelOutput cancel(CancelInput input) { public TerminateOutput terminate(TerminateInput input) { TerminateWorkflowExecutionRequest.Builder request = TerminateWorkflowExecutionRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) - .setIdentity(clientOptions.getIdentity()) + .setNamespace(namespace) + .setIdentity(identity) .setWorkflowExecution(input.getWorkflowExecution()); if (input.getReason() != null) { request.setReason(input.getReason()); @@ -746,11 +737,7 @@ public TerminateOutput terminate(TerminateInput input) { request.setFirstExecutionRunId(input.getFirstExecutionRunId()); } DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter(input.getWorkflowExecution()); Optional payloads = dataConverterWithWorkflowContext.toPayloads(input.getDetails()); payloads.ifPresent(request::setDetails); genericClient.terminate(request.build()); @@ -762,16 +749,14 @@ public DescribeWorkflowOutput describe(DescribeWorkflowInput input) { DescribeWorkflowExecutionResponse response = genericClient.describeWorkflowExecution( DescribeWorkflowExecutionRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) + .setNamespace(namespace) .setExecution(input.getWorkflowExecution()) .build()); DataConverter dataConverterWithWorkflowContext = - clientOptions - .getDataConverter() - .withContext( - new WorkflowSerializationContext( - clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); + workflowConverter( + response.getWorkflowExecutionInfo().getExecution(), + response.getWorkflowExecutionInfo().getType().getName()); return new DescribeWorkflowOutput( new WorkflowExecutionDescription(response, dataConverterWithWorkflowContext)); @@ -780,7 +765,7 @@ public DescribeWorkflowOutput describe(DescribeWorkflowInput input) { @Override public CountWorkflowOutput countWorkflows(CountWorkflowsInput input) { CountWorkflowExecutionsRequest.Builder req = - CountWorkflowExecutionsRequest.newBuilder().setNamespace(clientOptions.getNamespace()); + CountWorkflowExecutionsRequest.newBuilder().setNamespace(namespace); if (input.getQuery() != null) { req.setQuery(input.getQuery()); } @@ -792,12 +777,14 @@ public CountWorkflowOutput countWorkflows(CountWorkflowsInput input) { public ListWorkflowExecutionsOutput listWorkflowExecutions(ListWorkflowExecutionsInput input) { ListWorkflowExecutionIterator iterator = new ListWorkflowExecutionIterator( - input.getQuery(), clientOptions.getNamespace(), input.getPageSize(), genericClient); + input.getQuery(), namespace, input.getPageSize(), genericClient); iterator.init(); Iterator wrappedIterator = Iterators.transform( iterator, - info -> new WorkflowExecutionMetadata(info, clientOptions.getDataConverter())); + info -> + new WorkflowExecutionMetadata( + info, workflowConverter(info.getExecution(), info.getType().getName()))); // IMMUTABLE here means that "interference" (in Java Streams terms) to this spliterator is // impossible diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java new file mode 100644 index 0000000000..b76b4156c5 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java @@ -0,0 +1,44 @@ +package io.temporal.internal.client; + +import com.google.common.base.Strings; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.converter.DataConverter; +import io.temporal.internal.payload.storage.ExternalStorageDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.context.WorkflowSerializationContext; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import javax.annotation.Nullable; + +/** Supplies a {@link DataConverter} for clients. */ +final class WorkflowClientDataConverterFactory { + + private final String namespace; + private final DataConverter baseConverter; + private final boolean externalStorageConfigured; + + WorkflowClientDataConverterFactory( + WorkflowClientOptions clientOptions, @Nullable ExternalStorageRunner externalStorage) { + this.namespace = clientOptions.getNamespace(); + this.externalStorageConfigured = externalStorage != null; + this.baseConverter = + externalStorage == null + ? clientOptions.getDataConverter() + : new ExternalStorageDataConverter(clientOptions.getDataConverter(), externalStorage); + } + + DataConverter forWorkflow( + String workflowId, @Nullable String runId, @Nullable String workflowType) { + DataConverter converter = + baseConverter.withContext(new WorkflowSerializationContext(namespace, workflowId)); + if (!externalStorageConfigured) { + return converter; + } + return ((ExternalStorageDataConverter) converter) + .withStorageTarget( + new StorageDriverWorkflowInfo( + namespace, + Strings.emptyToNull(workflowId), + Strings.emptyToNull(runId), + Strings.emptyToNull(workflowType))); + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java index 74eb0a5e7d..327a5cb6d4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactory.java @@ -4,24 +4,31 @@ import io.temporal.activity.ManualActivityCompletionClient; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.common.converter.DataConverter; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import javax.annotation.Nonnull; import javax.annotation.Nullable; public interface ManualActivityCompletionClientFactory { - /** - * Create a {@link ManualActivityCompletionClientFactory} that emits simple {@link - * ManualActivityCompletionClientImpl} implementations - */ static ManualActivityCompletionClientFactory newFactory( @Nonnull WorkflowServiceStubs service, @Nonnull String namespace, @Nonnull String identity, @Nonnull DataConverter dataConverter) { + return newFactory(service, namespace, identity, dataConverter, null); + } + + static ManualActivityCompletionClientFactory newFactory( + @Nonnull WorkflowServiceStubs service, + @Nonnull String namespace, + @Nonnull String identity, + @Nonnull DataConverter dataConverter, + @Nullable ExternalStorageRunner externalStorage) { return new ManualActivityCompletionClientFactoryImpl( - service, namespace, identity, dataConverter); + service, namespace, identity, dataConverter, externalStorage); } ManualActivityCompletionClient getClient(@Nonnull byte[] taskToken, @Nonnull Scope metricsScope); @@ -31,6 +38,12 @@ ManualActivityCompletionClient getClient( @Nonnull Scope metricsScope, @Nullable ActivitySerializationContext activitySerializationContext); + ManualActivityCompletionClient getClient( + @Nonnull byte[] taskToken, + @Nonnull Scope metricsScope, + @Nullable ActivitySerializationContext activitySerializationContext, + @Nullable StorageDriverTargetInfo storageTarget); + ManualActivityCompletionClient getClient( @Nonnull WorkflowExecution execution, @Nonnull String activityId, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java index 6c8237401e..286d11902a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryImpl.java @@ -1,11 +1,15 @@ package io.temporal.internal.client.external; import com.google.common.base.Preconditions; +import com.google.common.base.Strings; import com.uber.m3.tally.Scope; import io.temporal.activity.ManualActivityCompletionClient; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.common.converter.DataConverter; +import io.temporal.internal.payload.storage.ActivityStorageTargets; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import java.util.Objects; import javax.annotation.Nonnull; @@ -16,16 +20,19 @@ class ManualActivityCompletionClientFactoryImpl implements ManualActivityComplet private final DataConverter dataConverter; private final String namespace; private final String identity; + private final @Nullable ExternalStorageRunner externalStorage; ManualActivityCompletionClientFactoryImpl( @Nonnull WorkflowServiceStubs service, @Nonnull String namespace, @Nonnull String identity, - @Nonnull DataConverter dataConverter) { + @Nonnull DataConverter dataConverter, + @Nullable ExternalStorageRunner externalStorage) { this.service = Objects.requireNonNull(service); this.namespace = Objects.requireNonNull(namespace); this.identity = Objects.requireNonNull(identity); this.dataConverter = Objects.requireNonNull(dataConverter); + this.externalStorage = externalStorage; } @Override @@ -39,6 +46,26 @@ public ManualActivityCompletionClient getClient( @Nonnull byte[] taskToken, @Nonnull Scope metricsScope, @Nullable ActivitySerializationContext activitySerializationContext) { + StorageDriverTargetInfo storageTarget = + activitySerializationContext == null + ? null + : ActivityStorageTargets.newBuilder(namespace) + .setActivity( + null, null, Strings.emptyToNull(activitySerializationContext.getActivityType())) + .setWorkflow( + Strings.emptyToNull(activitySerializationContext.getWorkflowId()), + null, + Strings.emptyToNull(activitySerializationContext.getWorkflowType())) + .build(); + return getClient(taskToken, metricsScope, activitySerializationContext, storageTarget); + } + + @Override + public ManualActivityCompletionClient getClient( + @Nonnull byte[] taskToken, + @Nonnull Scope metricsScope, + @Nullable ActivitySerializationContext activitySerializationContext, + @Nullable StorageDriverTargetInfo storageTarget) { Preconditions.checkNotNull(metricsScope, "metricsScope"); Preconditions.checkNotNull(taskToken, "taskToken"); Preconditions.checkArgument(taskToken.length > 0, "empty taskToken"); @@ -51,7 +78,9 @@ public ManualActivityCompletionClient getClient( taskToken, null, null, - activitySerializationContext); + activitySerializationContext, + storageTarget, + externalStorage); } @Override @@ -71,6 +100,16 @@ public ManualActivityCompletionClient getClient( Preconditions.checkNotNull(metricsScope, "metricsScope"); Preconditions.checkNotNull(execution, "execution"); Preconditions.checkNotNull(activityId, "activityId"); + String activityRunId = + execution.getWorkflowId().isEmpty() ? Strings.emptyToNull(execution.getRunId()) : null; + String activityType = + activitySerializationContext == null + ? null + : Strings.emptyToNull(activitySerializationContext.getActivityType()); + String workflowType = + activitySerializationContext == null + ? null + : Strings.emptyToNull(activitySerializationContext.getWorkflowType()); return new ManualActivityCompletionClientImpl( service, namespace, @@ -80,6 +119,14 @@ public ManualActivityCompletionClient getClient( null, execution, activityId, - activitySerializationContext); + activitySerializationContext, + ActivityStorageTargets.newBuilder(namespace) + .setActivity(activityId, activityRunId, activityType) + .setWorkflow( + Strings.emptyToNull(execution.getWorkflowId()), + Strings.emptyToNull(execution.getRunId()), + workflowType) + .build(), + externalStorage); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java index 0e68b107b5..6a479cdabd 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java @@ -4,6 +4,7 @@ import com.google.common.base.Preconditions; import com.google.protobuf.ByteString; +import com.google.protobuf.Message; import com.uber.m3.tally.Scope; import io.grpc.Status; import io.grpc.StatusRuntimeException; @@ -12,12 +13,15 @@ import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.workflowservice.v1.*; import io.temporal.client.*; +import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.failure.CanceledFailure; import io.temporal.internal.client.ActivityClientHelper; import io.temporal.internal.common.OptionsUtils; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.RpcRetryOptions; import io.temporal.serviceclient.WorkflowServiceStubs; import java.util.Optional; @@ -41,6 +45,8 @@ class ManualActivityCompletionClientImpl implements ManualActivityCompletionClie private final byte[] taskToken; private final GrpcRetryer grpcRetryer; private final GrpcRetryer.GrpcRetryerOptions replyGrpcRetryerOptions; + private final @Nullable StorageDriverTargetInfo storageTarget; + private final @Nullable ExternalStorageRunner externalStorage; ManualActivityCompletionClientImpl( @Nonnull WorkflowServiceStubs service, @@ -51,8 +57,12 @@ class ManualActivityCompletionClientImpl implements ManualActivityCompletionClie @Nullable byte[] taskToken, @Nullable WorkflowExecution execution, @Nullable String activityId, - @Nullable ActivitySerializationContext context) { + @Nullable ActivitySerializationContext context, + @Nullable StorageDriverTargetInfo storageTarget, + @Nullable ExternalStorageRunner externalStorage) { this.service = service; + this.externalStorage = externalStorage; + this.storageTarget = storageTarget; this.dataConverterWithActivityExecutionContext = context != null ? dataConverter.withContext(context) : dataConverter; this.namespace = namespace; @@ -75,23 +85,35 @@ class ManualActivityCompletionClientImpl implements ManualActivityCompletionClie this.activityId = activityId; } + private T storeOutbound(T request) { + if (externalStorage == null) { + return request; + } + Message.Builder builder = request.toBuilder(); + externalStorage.store(builder, storageTarget, null, CancellationToken.none()); + @SuppressWarnings("unchecked") + T stored = (T) builder.build(); + return stored; + } + @Override public void complete(@Nullable Object result) { Optional payloads = dataConverterWithActivityExecutionContext.toPayloads(result); if (taskToken != null) { - RespondActivityTaskCompletedRequest.Builder request = + RespondActivityTaskCompletedRequest.Builder builder = RespondActivityTaskCompletedRequest.newBuilder() .setNamespace(namespace) .setIdentity(identity) .setTaskToken(ByteString.copyFrom(taskToken)); - payloads.ifPresent(request::setResult); + payloads.ifPresent(builder::setResult); try { + RespondActivityTaskCompletedRequest request = storeOutbound(builder.build()); grpcRetryer.retry( () -> service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .respondActivityTaskCompleted(request.build()), + .respondActivityTaskCompleted(request), replyGrpcRetryerOptions); } catch (Exception e) { processException(e); @@ -100,20 +122,21 @@ public void complete(@Nullable Object result) { if (activityId == null) { throw new IllegalArgumentException("Either activity id or task token are required"); } - RespondActivityTaskCompletedByIdRequest.Builder request = + RespondActivityTaskCompletedByIdRequest.Builder builder = RespondActivityTaskCompletedByIdRequest.newBuilder() .setActivityId(activityId) .setNamespace(namespace) .setWorkflowId(execution.getWorkflowId()) .setRunId(execution.getRunId()); - payloads.ifPresent(request::setResult); + payloads.ifPresent(builder::setResult); try { + RespondActivityTaskCompletedByIdRequest request = storeOutbound(builder.build()); grpcRetryer.retry( () -> service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .respondActivityTaskCompletedById(request.build()), + .respondActivityTaskCompletedById(request), replyGrpcRetryerOptions); } catch (Exception e) { processException(e); @@ -126,13 +149,13 @@ public void fail(@Nonnull Throwable exception) { Preconditions.checkNotNull(exception, "null exception"); // When converting failures reason is class name, details are serialized exception. if (taskToken != null) { - RespondActivityTaskFailedRequest request = + RespondActivityTaskFailedRequest.Builder builder = RespondActivityTaskFailedRequest.newBuilder() .setFailure(dataConverterWithActivityExecutionContext.exceptionToFailure(exception)) .setNamespace(namespace) - .setTaskToken(ByteString.copyFrom(taskToken)) - .build(); + .setTaskToken(ByteString.copyFrom(taskToken)); try { + RespondActivityTaskFailedRequest request = storeOutbound(builder.build()); grpcRetryer.retry( () -> service @@ -152,15 +175,15 @@ public void fail(@Nonnull Throwable exception) { if (activityId == null) { throw new IllegalArgumentException("Either activity id or task token are required"); } - RespondActivityTaskFailedByIdRequest request = + RespondActivityTaskFailedByIdRequest.Builder builder = RespondActivityTaskFailedByIdRequest.newBuilder() .setFailure(dataConverterWithActivityExecutionContext.exceptionToFailure(exception)) .setNamespace(namespace) .setWorkflowId(execution.getWorkflowId()) .setRunId(execution.getRunId()) - .setActivityId(activityId) - .build(); + .setActivityId(activityId); try { + RespondActivityTaskFailedByIdRequest request = storeOutbound(builder.build()); grpcRetryer.retry( () -> service @@ -177,15 +200,17 @@ public void fail(@Nonnull Throwable exception) { @Override public void recordHeartbeat(@Nullable Object details) throws CanceledFailure { try { + Optional payloads = dataConverterWithActivityExecutionContext.toPayloads(details); if (taskToken != null) { + RecordActivityTaskHeartbeatRequest.Builder builder = + RecordActivityTaskHeartbeatRequest.newBuilder() + .setNamespace(namespace) + .setIdentity(identity) + .setTaskToken(ByteString.copyFrom(taskToken)); + payloads.ifPresent(builder::setDetails); RecordActivityTaskHeartbeatResponse status = ActivityClientHelper.sendHeartbeatRequest( - service, - namespace, - identity, - taskToken, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope); + service, storeOutbound(builder.build()), metricsScope); if (status.getCancelRequested()) { throw new ActivityCanceledException(); } else if (status.getActivityReset()) { @@ -194,15 +219,17 @@ public void recordHeartbeat(@Nullable Object details) throws CanceledFailure { throw new ActivityPausedException(); } } else { + RecordActivityTaskHeartbeatByIdRequest.Builder builder = + RecordActivityTaskHeartbeatByIdRequest.newBuilder() + .setNamespace(namespace) + .setIdentity(identity) + .setWorkflowId(execution.getWorkflowId()) + .setRunId(execution.getRunId()) + .setActivityId(activityId); + payloads.ifPresent(builder::setDetails); RecordActivityTaskHeartbeatByIdResponse status = ActivityClientHelper.recordActivityTaskHeartbeatById( - service, - namespace, - identity, - execution, - activityId, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope); + service, storeOutbound(builder.build()), metricsScope); if (status.getCancelRequested()) { throw new ActivityCanceledException(); } else if (status.getActivityReset()) { @@ -221,18 +248,19 @@ public void reportCancellation(@Nullable Object details) { Optional convertedDetails = dataConverterWithActivityExecutionContext.toPayloads(details); if (taskToken != null) { - RespondActivityTaskCanceledRequest.Builder request = + RespondActivityTaskCanceledRequest.Builder builder = RespondActivityTaskCanceledRequest.newBuilder() .setNamespace(namespace) .setTaskToken(ByteString.copyFrom(taskToken)); - convertedDetails.ifPresent(request::setDetails); + convertedDetails.ifPresent(builder::setDetails); try { + RespondActivityTaskCanceledRequest request = storeOutbound(builder.build()); grpcRetryer.retry( () -> service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .respondActivityTaskCanceled(request.build()), + .respondActivityTaskCanceled(request), replyGrpcRetryerOptions); } catch (Exception e) { // There is nothing that can be done at this point. @@ -243,20 +271,21 @@ public void reportCancellation(@Nullable Object details) { if (activityId == null) { throw new IllegalArgumentException("Either activity id or task token are required"); } - RespondActivityTaskCanceledByIdRequest.Builder request = + RespondActivityTaskCanceledByIdRequest.Builder builder = RespondActivityTaskCanceledByIdRequest.newBuilder() .setNamespace(namespace) .setWorkflowId(execution.getWorkflowId()) .setRunId(OptionsUtils.safeGet(execution.getRunId())) .setActivityId(activityId); - convertedDetails.ifPresent(request::setDetails); + convertedDetails.ifPresent(builder::setDetails); try { + RespondActivityTaskCanceledByIdRequest request = storeOutbound(builder.build()); grpcRetryer.retry( () -> service .blockingStub() .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .respondActivityTaskCanceledById(request.build()), + .respondActivityTaskCanceledById(request), replyGrpcRetryerOptions); } catch (Exception e) { // There is nothing that can be done at this point. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ActivityStorageTargets.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ActivityStorageTargets.java new file mode 100644 index 0000000000..a8cda4b4e9 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ActivityStorageTargets.java @@ -0,0 +1,61 @@ +package io.temporal.internal.payload.storage; + +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import javax.annotation.Nullable; + +/** Chooses the storage target an activity's payloads belong to. */ +public final class ActivityStorageTargets { + + public static Builder newBuilder(String namespace) { + return new Builder(namespace); + } + + private ActivityStorageTargets() {} + + public static final class Builder { + private final String namespace; + private @Nullable String activityId; + private @Nullable String activityRunId; + private @Nullable String activityType; + private @Nullable String workflowId; + private @Nullable String workflowRunId; + private @Nullable String workflowType; + + private Builder(String namespace) { + this.namespace = namespace; + } + + public Builder setActivity( + @Nullable String activityId, + @Nullable String activityRunId, + @Nullable String activityType) { + this.activityId = activityId; + this.activityRunId = activityRunId; + this.activityType = activityType; + return this; + } + + public Builder setWorkflow( + @Nullable String workflowId, + @Nullable String workflowRunId, + @Nullable String workflowType) { + this.workflowId = workflowId; + this.workflowRunId = workflowRunId; + this.workflowType = workflowType; + return this; + } + + /** + * An activity scheduled by a workflow targets that workflow; a standalone activity targets + * itself. A workflow id is present only in the former case. + */ + public StorageDriverTargetInfo build() { + if (workflowId != null) { + return new StorageDriverWorkflowInfo(namespace, workflowId, workflowRunId, workflowType); + } + return new StorageDriverActivityInfo(namespace, activityId, activityRunId, activityType); + } + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java index ff528d46b3..47fbb92cd0 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java @@ -2,7 +2,9 @@ import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY; +import com.google.common.base.Strings; import com.google.protobuf.ByteString; +import com.google.protobuf.Message; import com.uber.m3.tally.Scope; import com.uber.m3.tally.Stopwatch; import com.uber.m3.util.Duration; @@ -10,11 +12,16 @@ import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributesOrBuilder; import io.temporal.api.common.v1.WorkflowExecution; import io.temporal.api.workflowservice.v1.*; +import io.temporal.failure.ApplicationFailure; import io.temporal.internal.activity.ActivityPollResponseToInfo; import io.temporal.internal.common.ProtobufTimeUtils; +import io.temporal.internal.concurrent.structured.CancelSource; import io.temporal.internal.logging.LoggerTag; +import io.temporal.internal.payload.storage.ActivityStorageTargets; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.internal.worker.ActivityTaskHandler.Result; +import io.temporal.payload.storage.StorageDriverTargetInfo; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.rpcretry.DefaultStubServiceOperationRpcRetryOptions; @@ -24,9 +31,11 @@ import io.temporal.worker.tuning.PollerBehaviorAutoscaling; import java.util.Objects; import java.util.Optional; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; @@ -52,6 +61,9 @@ final class ActivityWorker implements SuspendableWorker { private final PollerTracker pollerTracker; private final NamespaceCapabilities namespaceCapabilities; + final CancelSource storageCancellation = + new CancelSource<>(() -> new CancellationException("Worker shutdown")); + public ActivityWorker( @Nonnull WorkflowServiceStubs service, @Nonnull String namespace, @@ -159,6 +171,9 @@ private String workerControlTaskQueue() { @Override public CompletableFuture shutdown(ShutdownManager shutdownManager, boolean interruptTasks) { + if (interruptTasks) { + storageCancellation.cancel(); + } String supplierName = this + "#executorSlots"; return poller .shutdown(shutdownManager, interruptTasks) @@ -257,6 +272,27 @@ public String toString() { options.getIdentity(), namespace, taskQueue); } + static StorageDriverTargetInfo storageTargetForActivityTask( + String namespace, PollActivityTaskQueueResponseOrBuilder pollResponse) { + WorkflowExecution execution = pollResponse.getWorkflowExecution(); + return ActivityStorageTargets.newBuilder(namespace) + .setActivity( + pollResponse.getActivityId(), + Strings.emptyToNull(pollResponse.getActivityRunId()), + pollResponse.getActivityType().getName()) + .setWorkflow( + Strings.emptyToNull(execution.getWorkflowId()), + Strings.emptyToNull(execution.getRunId()), + pollResponse.getWorkflowType().getName()) + .build(); + } + + private static final class ExternalStorageTaskFailure extends RuntimeException { + ExternalStorageTaskFailure(String message, Throwable cause) { + super(message, cause); + } + } + private class TaskHandlerImpl implements PollTaskExecutor.TaskHandler { final ActivityTaskHandler handler; @@ -331,6 +367,7 @@ public void handle(ActivityTask task) throws Exception { } private ActivityTaskHandler.Result handleActivity(ActivityTask task, Scope metricsScope) { + task = retrieveInboundPayloads(task); PollActivityTaskQueueResponseOrBuilder pollResponse = task.getResponse(); ByteString taskToken = pollResponse.getTaskToken(); metricsScope @@ -354,7 +391,10 @@ private ActivityTaskHandler.Result handleActivity(ActivityTask task, Scope metri } try { - sendReply(taskToken, result, metricsScope); + sendReply(taskToken, result, metricsScope, activityStorageTarget(pollResponse)); + } catch (ExternalStorageTaskFailure e) { + sendStorageFailure(taskToken, pollResponse, metricsScope, e.getCause()); + return result; } catch (Exception e) { logExceptionDuringResultReporting(e, pollResponse, result); // TODO this class doesn't report activity success and failure metrics now, instead it's @@ -392,16 +432,20 @@ public Throwable wrapFailure(ActivityTask t, Throwable failure) { // TODO: Suppress warning until the SDK supports deployment @SuppressWarnings("deprecation") private void sendReply( - ByteString taskToken, ActivityTaskHandler.Result response, Scope metricsScope) { + ByteString taskToken, + ActivityTaskHandler.Result response, + Scope metricsScope, + @Nullable StorageDriverTargetInfo storageTarget) { RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted(); if (taskCompleted != null) { - RespondActivityTaskCompletedRequest request = + RespondActivityTaskCompletedRequest.Builder completedBuilder = taskCompleted.toBuilder() .setTaskToken(taskToken) .setIdentity(options.getIdentity()) .setNamespace(namespace) - .setWorkerVersion(options.workerVersionStamp()) - .build(); + .setWorkerVersion(options.workerVersionStamp()); + storeOutboundPayloads(completedBuilder, storageTarget); + RespondActivityTaskCompletedRequest request = completedBuilder.build(); grpcRetryer.retry( () -> @@ -413,13 +457,14 @@ private void sendReply( } else { Result.TaskFailedResult taskFailed = response.getTaskFailed(); if (taskFailed != null) { - RespondActivityTaskFailedRequest request = + RespondActivityTaskFailedRequest.Builder failedBuilder = taskFailed.getTaskFailedRequest().toBuilder() .setTaskToken(taskToken) .setIdentity(options.getIdentity()) .setNamespace(namespace) - .setWorkerVersion(options.workerVersionStamp()) - .build(); + .setWorkerVersion(options.workerVersionStamp()); + storeOutboundPayloads(failedBuilder, storageTarget); + RespondActivityTaskFailedRequest request = failedBuilder.build(); grpcRetryer.retry( () -> @@ -431,13 +476,14 @@ private void sendReply( } else { RespondActivityTaskCanceledRequest taskCanceled = response.getTaskCanceled(); if (taskCanceled != null) { - RespondActivityTaskCanceledRequest request = + RespondActivityTaskCanceledRequest.Builder canceledBuilder = taskCanceled.toBuilder() .setTaskToken(taskToken) .setIdentity(options.getIdentity()) .setNamespace(namespace) - .setWorkerVersion(options.workerVersionStamp()) - .build(); + .setWorkerVersion(options.workerVersionStamp()); + storeOutboundPayloads(canceledBuilder, storageTarget); + RespondActivityTaskCanceledRequest request = canceledBuilder.build(); grpcRetryer.retry( () -> @@ -452,6 +498,76 @@ private void sendReply( // Manual activity completion } + private ActivityTask retrieveInboundPayloads(ActivityTask task) { + ExternalStorageRunner externalStorageRunner = options.getExternalStorageRunner(); + PollActivityTaskQueueResponseOrBuilder response = task.getResponse(); + PollActivityTaskQueueResponse built = + response instanceof PollActivityTaskQueueResponse + ? (PollActivityTaskQueueResponse) response + : ((PollActivityTaskQueueResponse.Builder) response).build(); + if (externalStorageRunner == null) { + ExternalStorageRunner.throwIfContainsReference(built); + return task; + } + return new ActivityTask( + externalStorageRunner.retrieve(built, storageCancellation.token()), + task.getPermit(), + task.getCompletionCallback()); + } + + private void storeOutboundPayloads( + Message.Builder builder, @Nullable StorageDriverTargetInfo target) { + ExternalStorageRunner externalStorageRunner = options.getExternalStorageRunner(); + if (externalStorageRunner == null) { + return; + } + try { + externalStorageRunner.store(builder, target, null, storageCancellation.token()); + } catch (Throwable e) { + throw new ExternalStorageTaskFailure("External storage store failed", e); + } + } + + @SuppressWarnings("deprecation") + private void sendStorageFailure( + ByteString taskToken, + PollActivityTaskQueueResponseOrBuilder pollResponse, + Scope metricsScope, + Throwable e) { + log.warn("External storage failed for an activity task", e); + ApplicationFailure applicationFailure = + ApplicationFailure.newBuilder() + .setMessage("External storage failed: " + e.getMessage()) + .setType(ExternalStorageTaskFailure.class.getSimpleName()) + .build(); + applicationFailure.setStackTrace(new StackTraceElement[0]); + RespondActivityTaskFailedRequest.Builder failedBuilder = + RespondActivityTaskFailedRequest.newBuilder() + .setTaskToken(taskToken) + .setIdentity(options.getIdentity()) + .setNamespace(namespace) + .setWorkerVersion(options.workerVersionStamp()) + .setFailure(options.getDataConverter().exceptionToFailure(applicationFailure)); + storeOutboundPayloads(failedBuilder, activityStorageTarget(pollResponse)); + RespondActivityTaskFailedRequest request = failedBuilder.build(); + grpcRetryer.retry( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .respondActivityTaskFailed(request), + replyGrpcRetryerOptions); + } + + @Nullable + private StorageDriverTargetInfo activityStorageTarget( + PollActivityTaskQueueResponseOrBuilder pollResponse) { + if (options.getExternalStorageRunner() == null) { + return null; + } + return storageTargetForActivityTask(namespace, pollResponse); + } + private void logExceptionDuringResultReporting( Exception e, PollActivityTaskQueueResponseOrBuilder pollResponse, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java index 94d2f5dee3..d5cf77f135 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java @@ -59,7 +59,8 @@ public SyncActivityWorker( options.getMaxHeartbeatThrottleInterval(), options.getDefaultHeartbeatThrottleInterval(), options.getDataConverter(), - heartbeatExecutor); + heartbeatExecutor, + options.getExternalStorageRunner()); this.taskHandler = new ActivityTaskHandlerImpl( namespace, diff --git a/temporal-sdk/src/test/java/io/temporal/client/WorkflowExecutionMetadataTest.java b/temporal-sdk/src/test/java/io/temporal/client/WorkflowExecutionMetadataTest.java new file mode 100644 index 0000000000..f0fb9ed37f --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/WorkflowExecutionMetadataTest.java @@ -0,0 +1,103 @@ +package io.temporal.client; + +import static org.junit.Assert.assertEquals; + +import io.temporal.api.common.v1.Memo; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.workflow.v1.WorkflowExecutionInfo; +import io.temporal.common.CancellationToken; +import io.temporal.common.converter.DataConverter; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import org.junit.Test; + +public class WorkflowExecutionMetadataTest { + + @Test + public void getMemoResolvesAnExternalStorageReference() { + ExternalStorage config = + ExternalStorage.newBuilder() + .setDriver(new InMemoryDriver()) + .setPayloadSizeThreshold(0) + .build(); + DataConverter converter = DefaultDataConverter.newDefaultInstance(); + ExternalStorageRunner storage = ExternalStorageRunner.create(config); + + Payloads.Builder value = converter.toPayloads("big-memo").get().toBuilder(); + storage.store(value, null, null, CancellationToken.none()); + Payload reference = value.build().getPayloads(0); + WorkflowExecutionInfo info = + WorkflowExecutionInfo.newBuilder() + .setMemo(Memo.newBuilder().putFields("k", reference)) + .build(); + + WorkflowExecutionMetadata metadata = + new WorkflowExecutionMetadata(info, new ExternalStorageDataConverter(converter, storage)); + + assertEquals("big-memo", metadata.getMemo("k", String.class)); + } + + @Test + public void getMemoReadsAnInlineValueWithoutExternalStorage() { + DataConverter converter = DefaultDataConverter.newDefaultInstance(); + Payload inline = converter.toPayloads("plain").get().getPayloads(0); + WorkflowExecutionInfo info = + WorkflowExecutionInfo.newBuilder() + .setMemo(Memo.newBuilder().putFields("k", inline)) + .build(); + + WorkflowExecutionMetadata metadata = new WorkflowExecutionMetadata(info, converter); + + assertEquals("plain", metadata.getMemo("k", String.class)); + } + + private static final class InMemoryDriver implements StorageDriver { + private final Map objects = new HashMap<>(); + private int counter = 0; + + @Override + public String getName() { + return "test"; + } + + @Override + public String getType() { + return "test.inmemory"; + } + + @Override + public synchronized CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + List claims = new ArrayList<>(); + for (Payload payload : payloads) { + String key = "k-" + (counter++); + objects.put(key, payload); + claims.add(new StorageDriverClaim(Collections.singletonMap("key", key))); + } + return CompletableFuture.completedFuture(claims); + } + + @Override + public synchronized CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + List payloads = new ArrayList<>(); + for (StorageDriverClaim claim : claims) { + payloads.add(objects.get(claim.getClaimData().get("key"))); + } + return CompletableFuture.completedFuture(payloads); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java new file mode 100644 index 0000000000..8cff52cad2 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java @@ -0,0 +1,76 @@ +package io.temporal.internal.activity; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import com.uber.m3.tally.Scope; +import io.temporal.activity.ActivityInfo; +import io.temporal.activity.ManualActivityCompletionClient; +import io.temporal.client.WorkflowClient; +import io.temporal.common.converter.GlobalDataConverter; +import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.serviceclient.WorkflowServiceStubs; +import java.time.Duration; +import java.util.concurrent.ScheduledExecutorService; +import org.junit.Test; + +public class ActivityExecutionContextImplTest { + + @Test + public void localManualCompletionIncludesActivityTarget() { + WorkflowClient client = mock(WorkflowClient.class); + when(client.getWorkflowServiceStubs()).thenReturn(mock(WorkflowServiceStubs.class)); + ActivityInfo info = mock(ActivityInfo.class); + when(info.getNamespace()).thenReturn("test-namespace"); + when(info.getWorkflowId()).thenReturn(null); + when(info.getWorkflowType()).thenReturn(null); + when(info.getActivityId()).thenReturn("activity-id"); + when(info.getActivityRunId()).thenReturn("activity-run-id"); + when(info.getActivityType()).thenReturn("activity-type"); + when(info.getActivityTaskQueue()).thenReturn("task-queue"); + when(info.getTaskToken()).thenReturn(new byte[] {1, 2, 3}); + ManualActivityCompletionClientFactory completionClientFactory = + mock(ManualActivityCompletionClientFactory.class); + when(completionClientFactory.getClient( + any(byte[].class), + any(Scope.class), + any(ActivitySerializationContext.class), + any(StorageDriverTargetInfo.class))) + .thenReturn(mock(ManualActivityCompletionClient.class)); + NoopScope metricsScope = new NoopScope(); + ActivityExecutionContextImpl context = + new ActivityExecutionContextImpl( + client, + "test-namespace", + new Object(), + info, + GlobalDataConverter.get(), + mock(ScheduledExecutorService.class), + completionClientFactory, + () -> {}, + metricsScope, + "test-identity", + Duration.ofSeconds(60), + Duration.ofSeconds(30), + () -> {}, + null); + + context.useLocalManualCompletion(); + + verify(completionClientFactory) + .getClient( + eq(new byte[] {1, 2, 3}), + eq(metricsScope), + any(ActivitySerializationContext.class), + eq( + new StorageDriverActivityInfo( + "test-namespace", "activity-id", "activity-run-id", "activity-type"))); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java index 1379aed154..b9a10ac119 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/activity/HeartbeatContextImplTest.java @@ -18,6 +18,8 @@ import io.temporal.common.CancellationToken; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.failure.TimeoutFailure; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.testUtils.Eventually; import java.time.Duration; @@ -329,7 +331,8 @@ public void factoryCancelByTaskTokenCompletesCancellationToken() { Duration.ofSeconds(60), Duration.ofSeconds(30), GlobalDataConverter.get(), - heartbeatExecutor); + heartbeatExecutor, + null); ActivityInfoInternal info = activityInfoWithHeartbeatTimeout(Duration.ofSeconds(10)); InternalActivityExecutionContext context = @@ -363,6 +366,7 @@ private HeartbeatContextImpl createHeartbeatContext( "test-identity", maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, + null, TEST_BUFFER_MILLIS); } @@ -390,4 +394,29 @@ private static ActivityInfoInternal activityInfoWithHeartbeatTimeout(Duration he when(info.getCompletionHandle()).thenReturn(() -> {}); return info; } + + @Test + public void storageTargetForStandaloneActivityTargetsTheActivity() { + ActivityInfo info = mock(ActivityInfo.class); + when(info.getActivityRunId()).thenReturn("act-run-1"); + when(info.getActivityId()).thenReturn("act-1"); + when(info.getActivityType()).thenReturn("MyActivity"); + + assertEquals( + new StorageDriverActivityInfo("ns", "act-1", "act-run-1", "MyActivity"), + HeartbeatContextImpl.storageTargetForActivity("ns", info)); + } + + @Test + public void storageTargetForWorkflowActivityTargetsTheWorkflow() { + ActivityInfo info = mock(ActivityInfo.class); + when(info.getActivityRunId()).thenReturn(null); + when(info.getWorkflowId()).thenReturn("wf-1"); + when(info.getWorkflowRunId()).thenReturn("wf-run-1"); + when(info.getWorkflowType()).thenReturn("MyWorkflow"); + + assertEquals( + new StorageDriverWorkflowInfo("ns", "wf-1", "wf-run-1", "MyWorkflow"), + HeartbeatContextImpl.storageTargetForActivity("ns", info)); + } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java new file mode 100644 index 0000000000..48e7271ee5 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java @@ -0,0 +1,133 @@ +package io.temporal.internal.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionResponse; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.common.interceptors.Header; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowStartInput; +import io.temporal.internal.client.external.GenericWorkflowClient; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import org.junit.Test; + +public class RootWorkflowClientInvokerStorageTargetTest { + + private static final String NAMESPACE = "test-namespace"; + + @Test + public void startCarriesTheWorkflowTypeButNoRunIdYet() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + when(rpc.start(any())).thenReturn(StartWorkflowExecutionResponse.getDefaultInstance()); + + invoker(rpc, driver) + .start( + new WorkflowStartInput( + "wf-1", + "MyWorkflowType", + Header.empty(), + new Object[] {"argument"}, + WorkflowOptions.newBuilder().setTaskQueue("tq").build())); + + StorageDriverWorkflowInfo target = (StorageDriverWorkflowInfo) driver.lastTarget; + assertEquals(NAMESPACE, target.getNamespace()); + assertEquals("wf-1", target.getId()); + assertEquals("MyWorkflowType", target.getType()); + assertNull(target.getRunId()); + } + + @Test + public void signalCarriesTheRunId() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + + invoker(rpc, driver) + .signal( + new WorkflowSignalInput( + WorkflowExecution.newBuilder().setWorkflowId("wf-2").setRunId("run-9").build(), + "mySignal", + Header.empty(), + new Object[] {"argument"})); + + StorageDriverWorkflowInfo target = (StorageDriverWorkflowInfo) driver.lastTarget; + assertEquals("wf-2", target.getId()); + assertEquals("run-9", target.getRunId()); + } + + @Test + public void anAbsentRunIdArrivesAsNullNotEmptyString() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + + invoker(rpc, driver) + .signal( + new WorkflowSignalInput( + WorkflowExecution.newBuilder().setWorkflowId("wf-3").build(), + "mySignal", + Header.empty(), + new Object[] {"argument"})); + + assertNull(((StorageDriverWorkflowInfo) driver.lastTarget).getRunId()); + } + + private static RootWorkflowClientInvoker invoker( + GenericWorkflowClient rpc, StorageDriver driver) { + return new RootWorkflowClientInvoker( + rpc, + WorkflowClientOptions.newBuilder().setNamespace(NAMESPACE).validateAndBuildWithDefaults(), + new WorkerFactoryRegistry(), + ExternalStorageRunner.create( + ExternalStorage.newBuilder().setDriver(driver).setPayloadSizeThreshold(0).build())); + } + + private static final class CapturingDriver implements StorageDriver { + volatile StorageDriverTargetInfo lastTarget; + private int counter = 0; + + @Override + public String getName() { + return "test"; + } + + @Override + public String getType() { + return "test.capturing"; + } + + @Override + public synchronized CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + lastTarget = context.getTarget(); + List claims = new ArrayList<>(); + for (int i = 0; i < payloads.size(); i++) { + claims.add(new StorageDriverClaim(Collections.singletonMap("key", "k-" + (counter++)))); + } + return CompletableFuture.completedFuture(claims); + } + + @Override + public CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryStorageTargetTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryStorageTargetTest.java new file mode 100644 index 0000000000..08b2f0f295 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientFactoryStorageTargetTest.java @@ -0,0 +1,159 @@ +package io.temporal.internal.client.external; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import io.temporal.activity.ManualActivityCompletionClient; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import org.junit.Before; +import org.junit.Test; + +/** + * A workflow-scheduled activity must target its workflow no matter which completion entry point is + * used, so that manual completion and {@link io.temporal.internal.worker.ActivityWorker} select the + * same driver for the same activity. + */ +public class ManualActivityCompletionClientFactoryStorageTargetTest { + + private static final String NAMESPACE = "test-namespace"; + + private final CapturingDriver driver = new CapturingDriver(); + private ManualActivityCompletionClientFactoryImpl factory; + + @Before + public void setUp() { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); + when(service.getOptions()).thenReturn(WorkflowServiceStubsOptions.getDefaultInstance()); + factory = + new ManualActivityCompletionClientFactoryImpl( + service, + NAMESPACE, + "test-identity", + DefaultDataConverter.newDefaultInstance(), + ExternalStorageRunner.create( + ExternalStorage.newBuilder() + .setDriver(driver) + .setPayloadSizeThreshold(0) + .setMaxConcurrentPayloadVisits(1) + .build())); + } + + @Test + public void byIdWorkflowActivityTargetsItsWorkflow() { + StorageDriverTargetInfo target = + capture( + factory.getClient( + WorkflowExecution.newBuilder() + .setWorkflowId("workflow-id") + .setRunId("workflow-run-id") + .build(), + "activity-id", + new NoopScope(), + serializationContext())); + + assertEquals( + new StorageDriverWorkflowInfo(NAMESPACE, "workflow-id", "workflow-run-id", "workflow-type"), + target); + } + + @Test + public void byIdStandaloneActivityTargetsItself() { + StorageDriverTargetInfo target = + capture( + factory.getClient( + WorkflowExecution.newBuilder().setRunId("activity-run-id").build(), + "activity-id", + new NoopScope(), + serializationContext())); + + assertEquals( + new StorageDriverActivityInfo(NAMESPACE, "activity-id", "activity-run-id", "activity-type"), + target); + } + + @Test + public void taskTokenWorkflowActivityTargetsItsWorkflow() { + StorageDriverTargetInfo target = + capture(factory.getClient(new byte[] {1, 2, 3}, new NoopScope(), serializationContext())); + + assertEquals( + new StorageDriverWorkflowInfo(NAMESPACE, "workflow-id", null, "workflow-type"), target); + } + + @Test + public void taskTokenStandaloneActivityTargetsItself() { + ActivitySerializationContext standalone = + new ActivitySerializationContext(NAMESPACE, "", "", "activity-type", "task-queue", false); + + StorageDriverTargetInfo target = + capture(factory.getClient(new byte[] {1, 2, 3}, new NoopScope(), standalone)); + + assertEquals(new StorageDriverActivityInfo(NAMESPACE, null, null, "activity-type"), target); + } + + private static ActivitySerializationContext serializationContext() { + return new ActivitySerializationContext( + NAMESPACE, "workflow-id", "workflow-type", "activity-type", "task-queue", false); + } + + /** + * The driver records the target then fails, so completion aborts before any RPC and the test + * needs no service response. + */ + private StorageDriverTargetInfo capture(ManualActivityCompletionClient client) { + driver.lastTarget = null; + assertThrows(RuntimeException.class, () -> client.complete("result")); + return driver.lastTarget; + } + + private static final class CapturingDriver implements StorageDriver { + volatile StorageDriverTargetInfo lastTarget; + + @Override + public String getName() { + return "capturing"; + } + + @Override + public String getType() { + return "test.capturing"; + } + + @Override + public CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + lastTarget = context.getTarget(); + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new RuntimeException("storage failed")); + return failed; + } + + @Override + public CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java new file mode 100644 index 0000000000..31ac1a6046 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java @@ -0,0 +1,146 @@ +package io.temporal.internal.client.external; + +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.client.ActivityCompletionFailureException; +import io.temporal.common.converter.DefaultDataConverter; +import io.temporal.failure.ApplicationFailure; +import io.temporal.internal.payload.storage.ExternalStorageRunner; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import org.junit.Before; +import org.junit.Test; + +public class ManualActivityCompletionClientImplTest { + private final RuntimeException storageFailure = new RuntimeException("storage failed"); + private WorkflowServiceStubs service; + private ExternalStorageRunner externalStorage; + + @Before + public void setUp() { + service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); + when(service.getOptions()).thenReturn(WorkflowServiceStubsOptions.getDefaultInstance()); + externalStorage = + ExternalStorageRunner.create( + ExternalStorage.newBuilder() + .setDriver(new FailingDriver()) + .setPayloadSizeThreshold(0) + .setMaxConcurrentPayloadVisits(1) + .build()); + } + + @Test + public void taskTokenCompletionWrapsStorageFailure() { + ManualActivityCompletionClientImpl client = taskTokenClient(); + + ActivityCompletionFailureException failure = + assertThrows(ActivityCompletionFailureException.class, () -> client.complete("result")); + + assertSame(storageFailure, failure.getCause()); + verify(service, never()).blockingStub(); + } + + @Test + public void byIdFailureWrapsStorageFailure() { + ManualActivityCompletionClientImpl client = byIdClient(); + + ActivityCompletionFailureException failure = + assertThrows( + ActivityCompletionFailureException.class, + () -> client.fail(ApplicationFailure.newFailure("activity failed", "test", "details"))); + + assertSame(storageFailure, failure.getCause()); + verify(service, never()).blockingStub(); + } + + @Test + public void taskTokenCancellationIgnoresStorageFailure() { + taskTokenClient().reportCancellation("details"); + + verify(service, never()).blockingStub(); + } + + @Test + public void byIdCancellationIgnoresStorageFailure() { + byIdClient().reportCancellation("details"); + + verify(service, never()).blockingStub(); + } + + private ManualActivityCompletionClientImpl taskTokenClient() { + return new ManualActivityCompletionClientImpl( + service, + "test-namespace", + "test-identity", + DefaultDataConverter.newDefaultInstance(), + new NoopScope(), + new byte[] {1, 2, 3}, + null, + null, + null, + new StorageDriverActivityInfo( + "test-namespace", "activity-id", "activity-run-id", "activity-type"), + externalStorage); + } + + private ManualActivityCompletionClientImpl byIdClient() { + return new ManualActivityCompletionClientImpl( + service, + "test-namespace", + "test-identity", + DefaultDataConverter.newDefaultInstance(), + new NoopScope(), + null, + WorkflowExecution.newBuilder().setRunId("activity-run-id").build(), + "activity-id", + null, + new StorageDriverActivityInfo( + "test-namespace", "activity-id", "activity-run-id", "activity-type"), + externalStorage); + } + + private final class FailingDriver implements StorageDriver { + @Override + public String getName() { + return "test"; + } + + @Override + public String getType() { + return "test"; + } + + @Override + public CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + CompletableFuture> result = new CompletableFuture<>(); + result.completeExceptionally(storageFailure); + return result; + } + + @Override + public CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/testing/ActivityTestingExternalStorageTest.java b/temporal-sdk/src/test/java/io/temporal/internal/testing/ActivityTestingExternalStorageTest.java new file mode 100644 index 0000000000..b4a323710f --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/testing/ActivityTestingExternalStorageTest.java @@ -0,0 +1,125 @@ +package io.temporal.internal.testing; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import io.temporal.activity.Activity; +import io.temporal.activity.ActivityInterface; +import io.temporal.api.common.v1.Payload; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; +import io.temporal.testing.TestActivityEnvironment; +import io.temporal.testing.TestEnvironmentOptions; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.Timeout; + +public class ActivityTestingExternalStorageTest { + + private static final String DETAILS = "heartbeat-details"; + + public @Rule Timeout timeout = Timeout.seconds(10); + + private final InMemoryDriver driver = new InMemoryDriver(); + private TestActivityEnvironment testEnvironment; + + @Before + public void setUp() { + testEnvironment = + TestActivityEnvironment.newInstance( + TestEnvironmentOptions.newBuilder() + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setExternalStorage( + ExternalStorage.newBuilder() + .setDriver(driver) + .setPayloadSizeThreshold(0) + .build()) + .build()) + .build()); + } + + @After + public void tearDown() throws Exception { + testEnvironment.close(); + } + + @Test + public void theHeartbeatListenerSeesDetailsThatWereOffloaded() { + testEnvironment.registerActivitiesImplementations(new HeartbeatActivityImpl()); + AtomicReference observed = new AtomicReference<>(); + testEnvironment.setActivityHeartbeatListener(String.class, observed::set); + + String result = testEnvironment.newActivityStub(TestActivity.class).activity1("input"); + + assertEquals("input", result); + assertTrue("expected the heartbeat details to be offloaded", driver.stores.get() > 0); + assertEquals(DETAILS, observed.get()); + } + + @ActivityInterface + public interface TestActivity { + String activity1(String input); + } + + public static class HeartbeatActivityImpl implements TestActivity { + @Override + public String activity1(String input) { + Activity.getExecutionContext().heartbeat(DETAILS); + return input; + } + } + + private static final class InMemoryDriver implements StorageDriver { + private final Map objects = new HashMap<>(); + final AtomicInteger stores = new AtomicInteger(); + private int counter = 0; + + @Override + public String getName() { + return "test-heartbeat"; + } + + @Override + public String getType() { + return "test.inmemory"; + } + + @Override + public synchronized CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + stores.incrementAndGet(); + List claims = new ArrayList<>(); + for (Payload payload : payloads) { + String key = "obj-" + (counter++); + objects.put(key, payload); + claims.add(new StorageDriverClaim(Collections.singletonMap("key", key))); + } + return CompletableFuture.completedFuture(claims); + } + + @Override + public synchronized CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + List payloads = new ArrayList<>(); + for (StorageDriverClaim claim : claims) { + payloads.add(objects.get(claim.getClaimData().get("key"))); + } + return CompletableFuture.completedFuture(payloads); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerExternalStorageFailureTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerExternalStorageFailureTest.java new file mode 100644 index 0000000000..f23893b4c6 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerExternalStorageFailureTest.java @@ -0,0 +1,186 @@ +package io.temporal.internal.worker; + +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.activity.ActivityOptions; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.enums.v1.EventType; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.common.RetryOptions; +import io.temporal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriver; +import io.temporal.payload.storage.StorageDriverClaim; +import io.temporal.payload.storage.StorageDriverRetrieveContext; +import io.temporal.payload.storage.StorageDriverStoreContext; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +public class ActivityWorkerExternalStorageFailureTest { + + private static final String LARGE_RESULT = String.join("", Collections.nCopies(60, "0123456789")); + + private static final FlakyDriver driver = new FlakyDriver("activity-flaky"); + + private static final ExternalStorage storage = + ExternalStorage.newBuilder().setDriver(driver).setPayloadSizeThreshold(100).build(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(LargeResultWorkflowImpl.class) + .setActivityImplementations(new LargeResultActivityImpl()) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder().setExternalStorage(storage).build()) + .build(); + + @Before + public void resetState() { + driver.reset(); + LargeResultActivityImpl.attempts.set(0); + } + + @Test + public void aFailedOutboundStoreRetriesWithoutWaitingForTheActivityTimeout() { + String workflowId = "extstore-activity-" + UUID.randomUUID(); + driver.failStoresContaining.set(LARGE_RESULT); + + LargeResultWorkflow workflow = + testWorkflowRule + .getWorkflowClient() + .newWorkflowStub( + LargeResultWorkflow.class, + WorkflowOptions.newBuilder() + .setTaskQueue(testWorkflowRule.getTaskQueue()) + .setWorkflowId(workflowId) + .build()); + + Assert.assertEquals("ok", workflow.execute()); + Assert.assertEquals( + "expected exactly one injected store failure", 1, driver.injectedStoreFailures.get()); + Assert.assertEquals( + "expected the activity to run twice", 2, LargeResultActivityImpl.attempts.get()); + Assert.assertTrue( + "a reported failure must not leave an activity timeout in history", + testWorkflowRule + .getHistoryEvents(workflowId, EventType.EVENT_TYPE_ACTIVITY_TASK_TIMED_OUT) + .isEmpty()); + } + + @WorkflowInterface + public interface LargeResultWorkflow { + @WorkflowMethod + String execute(); + } + + @ActivityInterface + public interface LargeResultActivity { + @ActivityMethod + String run(); + } + + public static class LargeResultWorkflowImpl implements LargeResultWorkflow { + @Override + public String execute() { + LargeResultActivity activity = + Workflow.newActivityStub( + LargeResultActivity.class, + ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(60)) + .setRetryOptions( + RetryOptions.newBuilder() + .setInitialInterval(Duration.ofMillis(100)) + .setMaximumAttempts(3) + .build()) + .build()); + return activity.run(); + } + } + + public static class LargeResultActivityImpl implements LargeResultActivity { + static final AtomicInteger attempts = new AtomicInteger(); + + @Override + public String run() { + return attempts.incrementAndGet() == 1 ? LARGE_RESULT : "ok"; + } + } + + private static final class FlakyDriver implements StorageDriver { + private final String name; + private final Map objects = new HashMap<>(); + final AtomicReference failStoresContaining = new AtomicReference<>(); + final AtomicInteger injectedStoreFailures = new AtomicInteger(); + private int counter = 0; + + FlakyDriver(String name) { + this.name = name; + } + + synchronized void reset() { + objects.clear(); + failStoresContaining.set(null); + injectedStoreFailures.set(0); + } + + @Override + public String getName() { + return name; + } + + @Override + public String getType() { + return "test.activity.flaky"; + } + + @Override + public synchronized CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + String marker = failStoresContaining.get(); + if (marker != null) { + for (Payload payload : payloads) { + if (payload.getData().toStringUtf8().contains(marker)) { + failStoresContaining.set(null); + injectedStoreFailures.incrementAndGet(); + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new IllegalStateException("storage unavailable")); + return failed; + } + } + } + List claims = new ArrayList<>(); + for (Payload payload : payloads) { + String key = name + "-" + (counter++); + objects.put(key, payload); + claims.add(new StorageDriverClaim(Collections.singletonMap("key", key))); + } + return CompletableFuture.completedFuture(claims); + } + + @Override + public synchronized CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + List payloads = new ArrayList<>(); + for (StorageDriverClaim claim : claims) { + payloads.add(objects.get(claim.getClaimData().get("key"))); + } + return CompletableFuture.completedFuture(payloads); + } + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerTest.java new file mode 100644 index 0000000000..f82a04c65e --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerTest.java @@ -0,0 +1,88 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.common.v1.WorkflowType; +import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.tuning.SlotSupplier; +import org.junit.Test; + +public class ActivityWorkerTest { + + @Test + public void standaloneActivityTargetsTheActivity() { + PollActivityTaskQueueResponse response = + PollActivityTaskQueueResponse.newBuilder() + .setActivityId("act-1") + .setActivityRunId("run-1") + .setActivityType(ActivityType.newBuilder().setName("MyActivity")) + .build(); + + StorageDriverTargetInfo target = ActivityWorker.storageTargetForActivityTask("ns", response); + + assertEquals(new StorageDriverActivityInfo("ns", "act-1", "run-1", "MyActivity"), target); + } + + @Test + public void workflowActivityTargetsTheWorkflow() { + PollActivityTaskQueueResponse response = + PollActivityTaskQueueResponse.newBuilder() + .setActivityId("act-1") + .setActivityType(ActivityType.newBuilder().setName("MyActivity")) + .setWorkflowType(WorkflowType.newBuilder().setName("MyWorkflow")) + .setWorkflowExecution( + WorkflowExecution.newBuilder().setWorkflowId("wf-1").setRunId("wf-run-1")) + .build(); + + StorageDriverTargetInfo target = ActivityWorker.storageTargetForActivityTask("ns", response); + + assertEquals(new StorageDriverWorkflowInfo("ns", "wf-1", "wf-run-1", "MyWorkflow"), target); + } + + @Test + public void interruptingShutdownCancelsInFlightStorage() throws Exception { + ActivityWorker worker = worker(); + + worker.shutdown(new ShutdownManager(), true).get(); + + assertTrue(worker.storageCancellation.token().isCancellationRequested()); + } + + @Test + public void gracefulShutdownLeavesStorageRunning() throws Exception { + ActivityWorker worker = worker(); + + worker.shutdown(new ShutdownManager(), false).get(); + + assertFalse(worker.storageCancellation.token().isCancellationRequested()); + } + + @SuppressWarnings("unchecked") + private static ActivityWorker worker() { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn( + () -> + io.temporal.api.workflowservice.v1.GetSystemInfoResponse.Capabilities + .getDefaultInstance()); + return new ActivityWorker( + service, + "ns", + "tq", + 1.0, + SingleWorkerOptions.newBuilder().build(), + mock(ActivityTaskHandler.class), + mock(SlotSupplier.class), + mock(NamespaceCapabilities.class)); + } +} diff --git a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java index 1d6ebb92de..dc2a0d3d3c 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java +++ b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java @@ -30,7 +30,10 @@ import io.temporal.internal.activity.ActivityExecutionContextFactory; import io.temporal.internal.activity.ActivityExecutionContextFactoryImpl; import io.temporal.internal.activity.ActivityTaskHandlerImpl; +import io.temporal.internal.client.WorkflowClientInternal; import io.temporal.internal.common.ProtobufTimeUtils; +import io.temporal.internal.payload.storage.ExternalStorageDataConverter; +import io.temporal.internal.payload.storage.ExternalStorageRunner; import io.temporal.internal.sync.*; import io.temporal.internal.testservice.InProcessGRPCServer; import io.temporal.internal.worker.ActivityTask; @@ -77,6 +80,7 @@ public final class TestActivityEnvironmentInternal implements TestActivityEnviro private final TestEnvironmentOptions testEnvironmentOptions; private final WorkflowServiceStubs workflowServiceStubs; private final AtomicReference heartbeatDetails = new AtomicReference<>(); + private final DataConverter heartbeatDetailsConverter; private ClassConsumerPair activityHeartbeatListener; public TestActivityEnvironmentInternal(@Nullable TestEnvironmentOptions options) { @@ -100,16 +104,27 @@ public TestActivityEnvironmentInternal(@Nullable TestEnvironmentOptions options) this.workflowServiceStubs = WorkflowServiceStubs.newServiceStubs(serviceStubsOptionsBuilder.build()); + WorkflowClient client = + WorkflowClient.newInstance( + this.workflowServiceStubs, testEnvironmentOptions.getWorkflowClientOptions()); + ExternalStorageRunner externalStorageRunner = + ((WorkflowClientInternal) client.getInternal()).getExternalStorageRunner(); + DataConverter clientDataConverter = + testEnvironmentOptions.getWorkflowClientOptions().getDataConverter(); + this.heartbeatDetailsConverter = + externalStorageRunner == null + ? clientDataConverter + : new ExternalStorageDataConverter(clientDataConverter, externalStorageRunner); ActivityExecutionContextFactory activityExecutionContextFactory = new ActivityExecutionContextFactoryImpl( - WorkflowClient.newInstance( - this.workflowServiceStubs, testEnvironmentOptions.getWorkflowClientOptions()), + client, testEnvironmentOptions.getWorkflowClientOptions().getIdentity(), testEnvironmentOptions.getWorkflowClientOptions().getNamespace(), WorkerOptions.getDefaultInstance().getMaxHeartbeatThrottleInterval(), WorkerOptions.getDefaultInstance().getDefaultHeartbeatThrottleInterval(), - testEnvironmentOptions.getWorkflowClientOptions().getDataConverter(), - heartbeatExecutor); + clientDataConverter, + heartbeatExecutor, + externalStorageRunner); activityTaskHandler = new ActivityTaskHandlerImpl( testEnvironmentOptions.getWorkflowClientOptions().getNamespace(), @@ -138,14 +153,11 @@ public void recordActivityTaskHeartbeat( request.hasDetails() ? Optional.of(request.getDetails()) : Optional.empty(); Object details = - testEnvironmentOptions - .getWorkflowClientOptions() - .getDataConverter() - .fromPayloads( - 0, - requestDetails, - activityHeartbeatListener.valueClass, - activityHeartbeatListener.valueType); + heartbeatDetailsConverter.fromPayloads( + 0, + requestDetails, + activityHeartbeatListener.valueClass, + activityHeartbeatListener.valueType); activityHeartbeatListener.consumer.apply(details); } responseObserver.onNext(