From 7069249ec7d0ca9b9032e0e4a97e8cc8128ebdec Mon Sep 17 00:00:00 2001 From: Chris Constable Date: Wed, 19 Aug 2026 14:01:26 -0400 Subject: [PATCH] feat(extstore): integrate into activity worker, heartbeats, and client pipelines. --- .../client/WorkflowClientInternalImpl.java | 29 +- .../client/WorkflowClientOptions.java | 40 +- .../ActivityExecutionContextFactoryImpl.java | 12 +- .../ActivityExecutionContextImpl.java | 18 +- .../activity/HeartbeatContextImpl.java | 73 +++- .../internal/client/ActivityClientHelper.java | 36 +- .../client/RootWorkflowClientInvoker.java | 24 +- .../client/WorkflowClientInternal.java | 4 + .../ExternalStorageGenericWorkflowClient.java | 344 ++++++++++++++++++ ...ManualActivityCompletionClientFactory.java | 23 +- ...alActivityCompletionClientFactoryImpl.java | 40 +- .../ManualActivityCompletionClientImpl.java | 81 +++-- .../internal/worker/ActivityWorker.java | 104 +++++- .../internal/worker/SyncActivityWorker.java | 3 +- .../main/java/io/temporal/worker/Worker.java | 32 +- .../ActivityExecutionContextImplTest.java | 76 ++++ .../activity/HeartbeatContextImplTest.java | 31 +- ...kflowClientInvokerLinkPropagationTest.java | 3 +- ...ernalStorageGenericWorkflowClientTest.java | 129 +++++++ ...anualActivityCompletionClientImplTest.java | 146 ++++++++ .../internal/worker/ActivityWorkerTest.java | 45 +++ .../TestActivityEnvironmentInternal.java | 10 +- 22 files changed, 1177 insertions(+), 126 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClient.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/activity/ActivityExecutionContextImplTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClientTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerTest.java 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 e856ca4b46..e0b2152253 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java @@ -18,12 +18,15 @@ import io.temporal.internal.WorkflowThreadMarker; import io.temporal.internal.client.*; import io.temporal.internal.client.NexusStartWorkflowResponse; +import io.temporal.internal.client.external.ExternalStorageGenericWorkflowClient; import io.temporal.internal.client.external.GenericWorkflowClient; import io.temporal.internal.client.external.GenericWorkflowClientImpl; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; import io.temporal.internal.common.PluginUtils; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.internal.sync.StubMarker; import io.temporal.internal.worker.HeartbeatManager; +import io.temporal.payload.storage.ExternalStorageOptions; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.WorkflowServiceStubsPlugin; @@ -56,6 +59,7 @@ final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClient private final WorkerFactoryRegistry workerFactoryRegistry = new WorkerFactoryRegistry(); private final String workerGroupingKey = java.util.UUID.randomUUID().toString(); private final @Nullable HeartbeatManager heartbeatManager; + private final @Nullable ExternalStorage externalStorage; /** * Creates client that connects to an instance of the Temporal Service. Cannot be used from within @@ -106,7 +110,18 @@ public static WorkflowClient newInstance( .getOptions() .getMetricsScope() .tagged(MetricsTag.defaultTags(options.getNamespace())); - this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope); + ExternalStorageOptions externalStorageOptions = options.getExternalStorage(); + ExternalStorage externalStorage = + externalStorageOptions == null ? null : ExternalStorage.create(externalStorageOptions); + this.externalStorage = externalStorage; + GenericWorkflowClient genericClient = + new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope); + if (externalStorage != null) { + genericClient = + new ExternalStorageGenericWorkflowClient( + genericClient, externalStorage, options.getNamespace()); + } + this.genericClient = genericClient; this.interceptors = options.getInterceptors(); this.workflowClientCallsInvoker = initializeClientInvoker(); this.manualActivityCompletionClientFactory = @@ -114,7 +129,8 @@ public static WorkflowClient newInstance( workflowServiceStubs, options.getNamespace(), options.getIdentity(), - options.getDataConverter()); + options.getDataConverter(), + externalStorage); java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval(); if (!heartbeatInterval.isNegative()) { @@ -127,7 +143,8 @@ public static WorkflowClient newInstance( private WorkflowClientCallsInterceptor initializeClientInvoker() { WorkflowClientCallsInterceptor workflowClientInvoker = - new RootWorkflowClientInvoker(genericClient, options, workerFactoryRegistry); + new RootWorkflowClientInvoker( + genericClient, options, workerFactoryRegistry, externalStorage); for (WorkflowClientInterceptor clientInterceptor : interceptors) { workflowClientInvoker = clientInterceptor.workflowClientCallsInterceptor(workflowClientInvoker); @@ -815,6 +832,12 @@ public HeartbeatManager getHeartbeatManager() { return heartbeatManager; } + @Override + @Nullable + public ExternalStorage getExternalStorage() { + return externalStorage; + } + @Override public NexusStartWorkflowResponse startNexus( NexusStartWorkflowRequest request, Functions.Proc workflow) { diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java index e10defba51..2e41a995f2 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java @@ -7,12 +7,14 @@ import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.common.interceptors.WorkflowClientInterceptor; +import io.temporal.payload.storage.ExternalStorageOptions; import java.lang.management.ManagementFactory; import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Objects; +import javax.annotation.Nullable; /** Options for WorkflowClient configuration. */ public final class WorkflowClientOptions { @@ -52,6 +54,7 @@ public static final class Builder { private QueryRejectCondition queryRejectCondition; private WorkflowClientPlugin[] plugins; private Duration workerHeartbeatInterval; + private @Nullable ExternalStorageOptions externalStorage; private Builder() {} @@ -68,6 +71,7 @@ private Builder(WorkflowClientOptions options) { queryRejectCondition = options.queryRejectCondition; plugins = options.plugins; workerHeartbeatInterval = options.workerHeartbeatInterval; + externalStorage = options.externalStorage; } public Builder setNamespace(String namespace) { @@ -170,6 +174,16 @@ public Builder setWorkerHeartbeatInterval(Duration workerHeartbeatInterval) { return this; } + /** + * Configures offloading of large payloads to external storage for workflows and activities + * created through this client and its workers. + */ + @Experimental + public Builder setExternalStorage(@Nullable ExternalStorageOptions externalStorage) { + this.externalStorage = externalStorage; + return this; + } + public WorkflowClientOptions build() { return new WorkflowClientOptions( namespace, @@ -180,7 +194,8 @@ public WorkflowClientOptions build() { contextPropagators, queryRejectCondition, plugins == null ? EMPTY_PLUGINS : plugins, - resolveHeartbeatInterval(workerHeartbeatInterval)); + resolveHeartbeatInterval(workerHeartbeatInterval), + externalStorage); } /** @@ -207,7 +222,8 @@ public WorkflowClientOptions validateAndBuildWithDefaults() { ? QueryRejectCondition.QUERY_REJECT_CONDITION_UNSPECIFIED : queryRejectCondition, plugins == null ? EMPTY_PLUGINS : plugins, - resolveHeartbeatInterval(workerHeartbeatInterval)); + resolveHeartbeatInterval(workerHeartbeatInterval), + externalStorage); } private static Duration resolveHeartbeatInterval(Duration raw) { @@ -250,6 +266,8 @@ private static Duration resolveHeartbeatInterval(Duration raw) { private final Duration workerHeartbeatInterval; + private final @Nullable ExternalStorageOptions externalStorage; + private WorkflowClientOptions( String namespace, DataConverter dataConverter, @@ -259,7 +277,8 @@ private WorkflowClientOptions( List contextPropagators, QueryRejectCondition queryRejectCondition, WorkflowClientPlugin[] plugins, - Duration workerHeartbeatInterval) { + Duration workerHeartbeatInterval, + @Nullable ExternalStorageOptions externalStorage) { this.namespace = namespace; this.dataConverter = dataConverter; this.interceptors = interceptors; @@ -269,6 +288,7 @@ private WorkflowClientOptions( this.queryRejectCondition = queryRejectCondition; this.plugins = plugins; this.workerHeartbeatInterval = workerHeartbeatInterval; + this.externalStorage = externalStorage; } /** @@ -335,6 +355,12 @@ public Duration getWorkerHeartbeatInterval() { return workerHeartbeatInterval; } + /** External storage configuration, or null when external storage is disabled. */ + @Experimental + public @Nullable ExternalStorageOptions getExternalStorage() { + return externalStorage; + } + @Override public String toString() { return "WorkflowClientOptions{" @@ -359,6 +385,8 @@ public String toString() { + Arrays.toString(plugins) + ", workerHeartbeatInterval=" + workerHeartbeatInterval + + ", externalStorage=" + + externalStorage + '}'; } @@ -376,7 +404,8 @@ public boolean equals(Object o) { && queryRejectCondition == that.queryRejectCondition && Arrays.equals(plugins, that.plugins) && com.google.common.base.Objects.equal( - workerHeartbeatInterval, that.workerHeartbeatInterval); + workerHeartbeatInterval, that.workerHeartbeatInterval) + && com.google.common.base.Objects.equal(externalStorage, that.externalStorage); } @Override @@ -390,6 +419,7 @@ public int hashCode() { contextPropagators, queryRejectCondition, Arrays.hashCode(plugins), - workerHeartbeatInterval); + workerHeartbeatInterval, + externalStorage); } } 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..82df0fac4a 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.ExternalStorage; 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 ExternalStorage externalStorage; private final ConcurrentMap activeContexts = new ConcurrentHashMap<>(); @@ -31,7 +34,8 @@ public ActivityExecutionContextFactoryImpl( Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, DataConverter dataConverter, - ScheduledExecutorService heartbeatExecutor) { + ScheduledExecutorService heartbeatExecutor, + @Nullable ExternalStorage 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..8cee6b7a55 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,7 +10,9 @@ import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.external.ManualActivityCompletionClientFactory; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverActivityInfo; import io.temporal.workflow.Functions; import java.lang.reflect.Type; import java.time.Duration; @@ -18,6 +20,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 +58,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext { String identity, Duration maxHeartbeatThrottleInterval, Duration defaultHeartbeatThrottleInterval, - Functions.Proc closeCallback) { + Functions.Proc closeCallback, + @Nullable ExternalStorage externalStorage) { this.client = client; this.activity = activity; this.metricsScope = metricsScope; @@ -73,7 +77,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext { metricsScope, identity, maxHeartbeatThrottleInterval, - defaultHeartbeatThrottleInterval); + defaultHeartbeatThrottleInterval, + externalStorage); } /** @@ -155,7 +160,14 @@ public ManualActivityCompletionClient useLocalManualCompletion() { new ActivitySerializationContext(info); return new CompletionAwareManualCompletionClient( manualCompletionClientFactory.getClient( - info.getTaskToken(), metricsScope, activitySerializationContext), + info.getTaskToken(), + metricsScope, + activitySerializationContext, + new StorageDriverActivityInfo( + info.getNamespace(), + info.getActivityId(), + info.getActivityRunId(), + info.getActivityType())), 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..5e4dccc720 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,6 @@ package io.temporal.internal.activity; +import com.google.protobuf.ByteString; import com.uber.m3.tally.Scope; import io.grpc.Status; import io.grpc.StatusRuntimeException; @@ -7,6 +8,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 +16,22 @@ import io.temporal.failure.TimeoutFailure; import io.temporal.internal.client.ActivityClientHelper; import io.temporal.internal.concurrent.structured.CancelSource; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.payload.context.ActivitySerializationContext; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; 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 ExternalStorage 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 ExternalStorage 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 ExternalStorage externalStorage, long localHeartbeatTimeoutBufferMillis) { this.service = service; this.metricsScope = metricsScope; this.dataConverter = dataConverter; + this.externalStorage = externalStorage; this.dataConverterWithActivityContext = dataConverter.withContext( new ActivitySerializationContext( @@ -330,16 +343,60 @@ 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) { + String activityRunId = info.getActivityRunId(); + if (activityRunId != null) { + return new StorageDriverActivityInfo( + namespace, info.getActivityId(), activityRunId, info.getActivityType()); + } + return new StorageDriverWorkflowInfo( + namespace, info.getWorkflowId(), info.getWorkflowRunId(), info.getWorkflowType()); + } + + /** + * Offloads large heartbeat payloads aborting if the store call runs longer than the heartbeat + * interval or if the activity is cancelled. + */ + private RecordActivityTaskHeartbeatRequest offloadHeartbeat( + RecordActivityTaskHeartbeatRequest request) { + 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 { + return externalStorage.storeBlocking(request, activityStorageTarget(), 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); + RecordActivityTaskHeartbeatRequest request = builder.build(); + if (externalStorage != null) { + request = offloadHeartbeat(request); + } 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..6516c9d002 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 @@ -28,6 +28,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.ExternalStorage; +import io.temporal.internal.payload.storage.ExternalStorageResolvingDataConverter; import io.temporal.internal.worker.WorkerVersioningProtoUtils; import io.temporal.payload.context.WorkflowSerializationContext; import io.temporal.serviceclient.StatusUtils; @@ -50,15 +52,28 @@ public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor private final WorkflowClientOptions clientOptions; private final EagerWorkflowTaskDispatcher eagerWorkflowTaskDispatcher; private final WorkflowClientRequestFactory requestsHelper; + private final @Nullable ExternalStorage externalStorage; public RootWorkflowClientInvoker( GenericWorkflowClient genericClient, WorkflowClientOptions clientOptions, - WorkerFactoryRegistry workerFactoryRegistry) { + WorkerFactoryRegistry workerFactoryRegistry, + @Nullable ExternalStorage externalStorage) { this.genericClient = genericClient; this.clientOptions = clientOptions; this.eagerWorkflowTaskDispatcher = new EagerWorkflowTaskDispatcher(workerFactoryRegistry); this.requestsHelper = new WorkflowClientRequestFactory(clientOptions); + this.externalStorage = externalStorage; + } + + /** + * Wraps {@code base} so that reference payloads are resolved lazily when a result object's getter + * deserializes them. Returns {@code base} unchanged when external storage is not configured. + */ + private DataConverter lazyResolvingConverter(DataConverter base) { + return externalStorage == null + ? base + : new ExternalStorageResolvingDataConverter(base, externalStorage); } @Override @@ -774,7 +789,8 @@ public DescribeWorkflowOutput describe(DescribeWorkflowInput input) { clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId())); return new DescribeWorkflowOutput( - new WorkflowExecutionDescription(response, dataConverterWithWorkflowContext)); + new WorkflowExecutionDescription( + response, lazyResolvingConverter(dataConverterWithWorkflowContext))); } @Override @@ -797,7 +813,9 @@ public ListWorkflowExecutionsOutput listWorkflowExecutions(ListWorkflowExecution Iterator wrappedIterator = Iterators.transform( iterator, - info -> new WorkflowExecutionMetadata(info, clientOptions.getDataConverter())); + info -> + new WorkflowExecutionMetadata( + info, lazyResolvingConverter(clientOptions.getDataConverter()))); // 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/WorkflowClientInternal.java b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java index fc034a366b..d72dd275a8 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientInternal.java @@ -1,6 +1,7 @@ package io.temporal.internal.client; import io.temporal.client.WorkflowClient; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.internal.worker.HeartbeatManager; import io.temporal.worker.WorkerFactory; import io.temporal.workflow.Functions; @@ -25,4 +26,7 @@ public interface WorkflowClientInternal { @Nullable HeartbeatManager getHeartbeatManager(); + + @Nullable + ExternalStorage getExternalStorage(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClient.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClient.java new file mode 100644 index 0000000000..fc6cbeb323 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClient.java @@ -0,0 +1,344 @@ +package io.temporal.internal.client.external; + +import com.google.common.base.Strings; +import io.grpc.Deadline; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.*; +import io.temporal.internal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; +import java.util.concurrent.CompletableFuture; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +/** + * Decorates a {@link GenericWorkflowClient} to offload outbound request payloads to external + * storage and restore inbound response payloads. + * + *

Only constructed when external storage is configured, so {@code externalStorage} is never + * null. + */ +public final class ExternalStorageGenericWorkflowClient implements GenericWorkflowClient { + private final GenericWorkflowClient next; + private final ExternalStorage externalStorage; + private final String namespace; + + public ExternalStorageGenericWorkflowClient( + GenericWorkflowClient next, ExternalStorage externalStorage, String namespace) { + this.next = next; + this.externalStorage = externalStorage; + this.namespace = namespace; + } + + @Nullable + private StorageDriverTargetInfo workflowTarget(String workflowId, String runId, String type) { + return new StorageDriverWorkflowInfo( + namespace, + Strings.emptyToNull(workflowId), + Strings.emptyToNull(runId), + Strings.emptyToNull(type)); + } + + @Nullable + private StorageDriverTargetInfo workflowTarget(WorkflowExecution execution, String type) { + return workflowTarget(execution.getWorkflowId(), execution.getRunId(), type); + } + + @Nullable + private StorageDriverTargetInfo multiOperationTarget(ExecuteMultiOperationRequest request) { + for (ExecuteMultiOperationRequest.Operation operation : request.getOperationsList()) { + if (operation.hasStartWorkflow()) { + StartWorkflowExecutionRequest start = operation.getStartWorkflow(); + return workflowTarget(start.getWorkflowId(), null, start.getWorkflowType().getName()); + } + } + return null; + } + + @Override + public StartWorkflowExecutionResponse start(StartWorkflowExecutionRequest request) { + return next.start( + externalStorage.storeBlocking( + request, + workflowTarget(request.getWorkflowId(), null, request.getWorkflowType().getName()))); + } + + @Override + public SignalWorkflowExecutionResponse signal(SignalWorkflowExecutionRequest request) { + return next.signal( + externalStorage.storeBlocking( + request, workflowTarget(request.getWorkflowExecution(), null))); + } + + @Override + public SignalWithStartWorkflowExecutionResponse signalWithStart( + SignalWithStartWorkflowExecutionRequest request) { + return next.signalWithStart( + externalStorage.storeBlocking( + request, + workflowTarget(request.getWorkflowId(), null, request.getWorkflowType().getName()))); + } + + @Override + public void requestCancel(RequestCancelWorkflowExecutionRequest parameters) { + next.requestCancel(parameters); + } + + @Override + public QueryWorkflowResponse query(QueryWorkflowRequest queryParameters) { + QueryWorkflowRequest stored = + externalStorage.storeBlocking( + queryParameters, workflowTarget(queryParameters.getExecution(), null)); + return externalStorage.retrieveBlocking(next.query(stored)); + } + + @Override + public UpdateWorkflowExecutionResponse update( + @Nonnull UpdateWorkflowExecutionRequest updateParameters, @Nonnull Deadline deadline) { + UpdateWorkflowExecutionRequest stored = + externalStorage.storeBlocking( + updateParameters, workflowTarget(updateParameters.getWorkflowExecution(), null)); + return externalStorage.retrieveBlocking(next.update(stored, deadline)); + } + + @Override + public CompletableFuture pollUpdateAsync( + @Nonnull PollWorkflowExecutionUpdateRequest request, @Nonnull Deadline deadline) { + return next.pollUpdateAsync(request, deadline).thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public void terminate(TerminateWorkflowExecutionRequest request) { + next.terminate( + externalStorage.storeBlocking( + request, workflowTarget(request.getWorkflowExecution(), null))); + } + + @Override + public GetWorkflowExecutionHistoryResponse longPollHistory( + @Nonnull GetWorkflowExecutionHistoryRequest request, @Nonnull Deadline deadline) { + return externalStorage.retrieveBlocking(next.longPollHistory(request, deadline)); + } + + @Override + public CompletableFuture longPollHistoryAsync( + @Nonnull GetWorkflowExecutionHistoryRequest request, @Nonnull Deadline deadline) { + return next.longPollHistoryAsync(request, deadline) + .thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory( + @Nonnull GetWorkflowExecutionHistoryRequest request) { + return externalStorage.retrieveBlocking(next.getWorkflowExecutionHistory(request)); + } + + @Override + public CompletableFuture getWorkflowExecutionHistoryAsync( + @Nonnull GetWorkflowExecutionHistoryRequest request) { + return next.getWorkflowExecutionHistoryAsync(request) + .thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public ListWorkflowExecutionsResponse listWorkflowExecutions( + ListWorkflowExecutionsRequest listRequest) { + return next.listWorkflowExecutions(listRequest); + } + + @Override + public CompletableFuture listWorkflowExecutionsAsync( + ListWorkflowExecutionsRequest listRequest) { + return next.listWorkflowExecutionsAsync(listRequest); + } + + @Override + public CountWorkflowExecutionsResponse countWorkflowExecutions( + CountWorkflowExecutionsRequest request) { + return next.countWorkflowExecutions(request); + } + + @Override + public CreateScheduleResponse createSchedule(CreateScheduleRequest request) { + return next.createSchedule(externalStorage.storeBlocking(request, null)); + } + + @Override + public CompletableFuture listSchedulesAsync(ListSchedulesRequest request) { + return next.listSchedulesAsync(request).thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public UpdateScheduleResponse updateSchedule(UpdateScheduleRequest request) { + return next.updateSchedule(externalStorage.storeBlocking(request, null)); + } + + @Override + public PatchScheduleResponse patchSchedule(PatchScheduleRequest request) { + return next.patchSchedule(request); + } + + @Override + public DeleteScheduleResponse deleteSchedule(DeleteScheduleRequest request) { + return next.deleteSchedule(request); + } + + @Override + public DescribeScheduleResponse describeSchedule(DescribeScheduleRequest request) { + return externalStorage.retrieveBlocking(next.describeSchedule(request)); + } + + @Override + public DescribeWorkflowExecutionResponse describeWorkflowExecution( + DescribeWorkflowExecutionRequest request) { + return next.describeWorkflowExecution(request); + } + + @Override + public StartNexusOperationExecutionResponse startNexusOperationExecution( + @Nonnull StartNexusOperationExecutionRequest request) { + return next.startNexusOperationExecution(externalStorage.storeBlocking(request, null)); + } + + @Override + public DescribeNexusOperationExecutionResponse describeNexusOperationExecution( + @Nonnull DescribeNexusOperationExecutionRequest request) { + return externalStorage.retrieveBlocking(next.describeNexusOperationExecution(request)); + } + + @Override + public PollNexusOperationExecutionResponse pollNexusOperationExecution( + @Nonnull PollNexusOperationExecutionRequest request, @Nonnull Deadline deadline) { + return externalStorage.retrieveBlocking(next.pollNexusOperationExecution(request, deadline)); + } + + @Override + public CompletableFuture pollNexusOperationExecutionAsync( + @Nonnull PollNexusOperationExecutionRequest request, @Nonnull Deadline deadline) { + return next.pollNexusOperationExecutionAsync(request, deadline) + .thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public CompletableFuture listNexusOperationExecutionsAsync( + @Nonnull ListNexusOperationExecutionsRequest request) { + return next.listNexusOperationExecutionsAsync(request) + .thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public CountNexusOperationExecutionsResponse countNexusOperationExecutions( + @Nonnull CountNexusOperationExecutionsRequest request) { + return next.countNexusOperationExecutions(request); + } + + @Override + public RequestCancelNexusOperationExecutionResponse requestCancelNexusOperationExecution( + @Nonnull RequestCancelNexusOperationExecutionRequest request) { + return next.requestCancelNexusOperationExecution(request); + } + + @Override + public TerminateNexusOperationExecutionResponse terminateNexusOperationExecution( + @Nonnull TerminateNexusOperationExecutionRequest request) { + return next.terminateNexusOperationExecution(request); + } + + @Override + public DeleteNexusOperationExecutionResponse deleteNexusOperationExecution( + @Nonnull DeleteNexusOperationExecutionRequest request) { + return next.deleteNexusOperationExecution(request); + } + + @Override + @SuppressWarnings("deprecation") + public UpdateWorkerBuildIdCompatibilityResponse updateWorkerBuildIdCompatability( + UpdateWorkerBuildIdCompatibilityRequest request) { + return next.updateWorkerBuildIdCompatability(request); + } + + @Override + public ExecuteMultiOperationResponse executeMultiOperation( + ExecuteMultiOperationRequest request, @Nonnull Deadline deadline) { + ExecuteMultiOperationRequest stored = + externalStorage.storeBlocking(request, multiOperationTarget(request)); + return externalStorage.retrieveBlocking(next.executeMultiOperation(stored, deadline)); + } + + @Override + public StartActivityExecutionResponse startActivity(StartActivityExecutionRequest request) { + return next.startActivity( + externalStorage.storeBlocking( + request, + new StorageDriverActivityInfo( + namespace, + Strings.emptyToNull(request.getActivityId()), + null, + Strings.emptyToNull(request.getActivityType().getName())))); + } + + @Override + public PollActivityExecutionResponse pollActivity(PollActivityExecutionRequest request) { + return externalStorage.retrieveBlocking(next.pollActivity(request)); + } + + @Override + public PollActivityExecutionResponse pollActivity( + PollActivityExecutionRequest request, @Nonnull Deadline deadline) { + return externalStorage.retrieveBlocking(next.pollActivity(request, deadline)); + } + + @Override + public CompletableFuture pollActivityAsync( + PollActivityExecutionRequest request, @Nonnull Deadline deadline) { + return next.pollActivityAsync(request, deadline) + .thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public DescribeActivityExecutionResponse describeActivity( + DescribeActivityExecutionRequest request) { + return externalStorage.retrieveBlocking(next.describeActivity(request)); + } + + @Override + public void cancelActivity(RequestCancelActivityExecutionRequest request) { + next.cancelActivity(request); + } + + @Override + public void terminateActivity(TerminateActivityExecutionRequest request) { + next.terminateActivity(request); + } + + @Override + public ListActivityExecutionsResponse listActivities(ListActivityExecutionsRequest request) { + return externalStorage.retrieveBlocking(next.listActivities(request)); + } + + @Override + public CompletableFuture listActivitiesAsync( + ListActivityExecutionsRequest request) { + return next.listActivitiesAsync(request).thenComposeAsync(externalStorage::retrieveAsync); + } + + @Override + public CountActivityExecutionsResponse countActivities(CountActivityExecutionsRequest request) { + return next.countActivities(request); + } + + @Override + @SuppressWarnings("deprecation") + public GetWorkerBuildIdCompatibilityResponse getWorkerBuildIdCompatability( + GetWorkerBuildIdCompatibilityRequest req) { + return next.getWorkerBuildIdCompatability(req); + } + + @Override + @SuppressWarnings("deprecation") + public GetWorkerTaskReachabilityResponse GetWorkerTaskReachability( + GetWorkerTaskReachabilityRequest req) { + return next.GetWorkerTaskReachability(req); + } +} 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..e2c5e26178 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.ExternalStorage; 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 ExternalStorage 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..fa55983881 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.ExternalStorage; 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.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 ExternalStorage externalStorage; ManualActivityCompletionClientFactoryImpl( @Nonnull WorkflowServiceStubs service, @Nonnull String namespace, @Nonnull String identity, - @Nonnull DataConverter dataConverter) { + @Nonnull DataConverter dataConverter, + @Nullable ExternalStorage 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,23 @@ public ManualActivityCompletionClient getClient( @Nonnull byte[] taskToken, @Nonnull Scope metricsScope, @Nullable ActivitySerializationContext activitySerializationContext) { + StorageDriverTargetInfo storageTarget = + activitySerializationContext == null + ? null + : new StorageDriverActivityInfo( + namespace, + null, + null, + Strings.emptyToNull(activitySerializationContext.getActivityType())); + 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 +75,9 @@ public ManualActivityCompletionClient getClient( taskToken, null, null, - activitySerializationContext); + activitySerializationContext, + storageTarget, + externalStorage); } @Override @@ -71,6 +97,12 @@ 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()); return new ManualActivityCompletionClientImpl( service, namespace, @@ -80,6 +112,8 @@ public ManualActivityCompletionClient getClient( null, execution, activityId, - activitySerializationContext); + activitySerializationContext, + new StorageDriverActivityInfo(namespace, activityId, activityRunId, activityType), + 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..5617ab9ca8 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; @@ -16,8 +17,10 @@ import io.temporal.failure.CanceledFailure; import io.temporal.internal.client.ActivityClientHelper; import io.temporal.internal.common.OptionsUtils; +import io.temporal.internal.payload.storage.ExternalStorage; 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 +44,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 ExternalStorage externalStorage; ManualActivityCompletionClientImpl( @Nonnull WorkflowServiceStubs service, @@ -51,8 +56,12 @@ class ManualActivityCompletionClientImpl implements ManualActivityCompletionClie @Nullable byte[] taskToken, @Nullable WorkflowExecution execution, @Nullable String activityId, - @Nullable ActivitySerializationContext context) { + @Nullable ActivitySerializationContext context, + @Nullable StorageDriverTargetInfo storageTarget, + @Nullable ExternalStorage externalStorage) { this.service = service; + this.externalStorage = externalStorage; + this.storageTarget = storageTarget; this.dataConverterWithActivityExecutionContext = context != null ? dataConverter.withContext(context) : dataConverter; this.namespace = namespace; @@ -75,23 +84,30 @@ class ManualActivityCompletionClientImpl implements ManualActivityCompletionClie this.activityId = activityId; } + private T storeOutbound(T request) { + return externalStorage == null + ? request + : externalStorage.storeBlocking(request, storageTarget); + } + @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 +116,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 +143,14 @@ 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 unstoredRequest = RespondActivityTaskFailedRequest.newBuilder() .setFailure(dataConverterWithActivityExecutionContext.exceptionToFailure(exception)) .setNamespace(namespace) .setTaskToken(ByteString.copyFrom(taskToken)) .build(); try { + RespondActivityTaskFailedRequest request = storeOutbound(unstoredRequest); grpcRetryer.retry( () -> service @@ -152,7 +170,7 @@ public void fail(@Nonnull Throwable exception) { if (activityId == null) { throw new IllegalArgumentException("Either activity id or task token are required"); } - RespondActivityTaskFailedByIdRequest request = + RespondActivityTaskFailedByIdRequest unstoredRequest = RespondActivityTaskFailedByIdRequest.newBuilder() .setFailure(dataConverterWithActivityExecutionContext.exceptionToFailure(exception)) .setNamespace(namespace) @@ -161,6 +179,7 @@ public void fail(@Nonnull Throwable exception) { .setActivityId(activityId) .build(); try { + RespondActivityTaskFailedByIdRequest request = storeOutbound(unstoredRequest); grpcRetryer.retry( () -> service @@ -177,15 +196,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 +215,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 +244,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 +267,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/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java index ff528d46b3..dfb75c6e69 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 @@ -3,6 +3,7 @@ import static io.temporal.serviceclient.MetricsTag.METRICS_TAGS_CALL_OPTIONS_KEY; 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; @@ -13,8 +14,12 @@ import io.temporal.internal.activity.ActivityPollResponseToInfo; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.logging.LoggerTag; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.internal.worker.ActivityTaskHandler.Result; +import io.temporal.payload.storage.StorageDriverActivityInfo; +import io.temporal.payload.storage.StorageDriverTargetInfo; +import io.temporal.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.MetricsTag; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.serviceclient.rpcretry.DefaultStubServiceOperationRpcRetryOptions; @@ -27,6 +32,7 @@ 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; @@ -257,6 +263,24 @@ public String toString() { options.getIdentity(), namespace, taskQueue); } + static StorageDriverTargetInfo storageTargetForActivityTask( + String namespace, PollActivityTaskQueueResponseOrBuilder pollResponse) { + String activityRunId = pollResponse.getActivityRunId(); + if (!activityRunId.isEmpty()) { + return new StorageDriverActivityInfo( + namespace, + pollResponse.getActivityId(), + activityRunId, + pollResponse.getActivityType().getName()); + } + WorkflowExecution execution = pollResponse.getWorkflowExecution(); + return new StorageDriverWorkflowInfo( + namespace, + execution.getWorkflowId(), + execution.getRunId(), + pollResponse.getWorkflowType().getName()); + } + private class TaskHandlerImpl implements PollTaskExecutor.TaskHandler { final ActivityTaskHandler handler; @@ -331,6 +355,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 +379,7 @@ private ActivityTaskHandler.Result handleActivity(ActivityTask task, Scope metri } try { - sendReply(taskToken, result, metricsScope); + sendReply(taskToken, result, metricsScope, activityStorageTarget(pollResponse)); } catch (Exception e) { logExceptionDuringResultReporting(e, pollResponse, result); // TODO this class doesn't report activity success and failure metrics now, instead it's @@ -392,16 +417,21 @@ 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 = - taskCompleted.toBuilder() - .setTaskToken(taskToken) - .setIdentity(options.getIdentity()) - .setNamespace(namespace) - .setWorkerVersion(options.workerVersionStamp()) - .build(); + storeOutboundPayloads( + taskCompleted.toBuilder() + .setTaskToken(taskToken) + .setIdentity(options.getIdentity()) + .setNamespace(namespace) + .setWorkerVersion(options.workerVersionStamp()) + .build(), + storageTarget); grpcRetryer.retry( () -> @@ -414,12 +444,14 @@ private void sendReply( Result.TaskFailedResult taskFailed = response.getTaskFailed(); if (taskFailed != null) { RespondActivityTaskFailedRequest request = - taskFailed.getTaskFailedRequest().toBuilder() - .setTaskToken(taskToken) - .setIdentity(options.getIdentity()) - .setNamespace(namespace) - .setWorkerVersion(options.workerVersionStamp()) - .build(); + storeOutboundPayloads( + taskFailed.getTaskFailedRequest().toBuilder() + .setTaskToken(taskToken) + .setIdentity(options.getIdentity()) + .setNamespace(namespace) + .setWorkerVersion(options.workerVersionStamp()) + .build(), + storageTarget); grpcRetryer.retry( () -> @@ -432,12 +464,14 @@ private void sendReply( RespondActivityTaskCanceledRequest taskCanceled = response.getTaskCanceled(); if (taskCanceled != null) { RespondActivityTaskCanceledRequest request = - taskCanceled.toBuilder() - .setTaskToken(taskToken) - .setIdentity(options.getIdentity()) - .setNamespace(namespace) - .setWorkerVersion(options.workerVersionStamp()) - .build(); + storeOutboundPayloads( + taskCanceled.toBuilder() + .setTaskToken(taskToken) + .setIdentity(options.getIdentity()) + .setNamespace(namespace) + .setWorkerVersion(options.workerVersionStamp()) + .build(), + storageTarget); grpcRetryer.retry( () -> @@ -452,6 +486,36 @@ private void sendReply( // Manual activity completion } + private ActivityTask retrieveInboundPayloads(ActivityTask task) { + ExternalStorage externalStorage = options.getExternalStorage(); + PollActivityTaskQueueResponseOrBuilder response = task.getResponse(); + PollActivityTaskQueueResponse built = + response instanceof PollActivityTaskQueueResponse + ? (PollActivityTaskQueueResponse) response + : ((PollActivityTaskQueueResponse.Builder) response).build(); + if (externalStorage == null) { + ExternalStorage.throwIfContainsReference(built); + return task; + } + return new ActivityTask( + externalStorage.retrieveBlocking(built), task.getPermit(), task.getCompletionCallback()); + } + + private T storeOutboundPayloads( + T request, @Nullable StorageDriverTargetInfo target) { + ExternalStorage externalStorage = options.getExternalStorage(); + return externalStorage == null ? request : externalStorage.storeBlocking(request, target); + } + + @Nullable + private StorageDriverTargetInfo activityStorageTarget( + PollActivityTaskQueueResponseOrBuilder pollResponse) { + if (options.getExternalStorage() == 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..df449ca00b 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.getExternalStorage()); this.taskHandler = new ActivityTaskHandlerImpl( namespace, diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index b755134448..0752b26f16 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -22,6 +22,8 @@ import io.temporal.common.converter.DataConverter; import io.temporal.common.converter.EncodedValues; import io.temporal.failure.TemporalFailure; +import io.temporal.internal.client.WorkflowClientInternal; +import io.temporal.internal.payload.storage.ExternalStorage; import io.temporal.internal.sync.WorkflowInternal; import io.temporal.internal.sync.WorkflowThreadExecutor; import io.temporal.internal.worker.*; @@ -123,6 +125,8 @@ private static final class TaskSnapshot { this.options = WorkerOptions.newBuilder(options).validateAndBuildWithDefaults(); this.clientOptions = client.getOptions(); this.cache = cache; + ExternalStorage externalStorage = + ((WorkflowClientInternal) client.getInternal()).getExternalStorage(); factoryOptions = WorkerFactoryOptions.newBuilder(factoryOptions).validateAndBuildWithDefaults(); WorkflowClientOptions clientOptions = client.getOptions(); String namespace = clientOptions.getNamespace(); @@ -150,6 +154,7 @@ private static final class TaskSnapshot { taggedScope, workerInstanceKey, workerControlTaskQueue, + externalStorage, activityTaskAutoEnrollEligible); if (this.options.isLocalActivityWorkerOnly()) { activityWorker = null; @@ -185,6 +190,7 @@ private static final class TaskSnapshot { taggedScope, workerInstanceKey, workerControlTaskQueue, + externalStorage, nexusTaskAutoEnrollEligible); SlotSupplier nexusSlotSupplier = this.options.getWorkerTuner() == null @@ -206,6 +212,7 @@ private static final class TaskSnapshot { taggedScope, workerInstanceKey, workerControlTaskQueue, + externalStorage, workflowTaskAutoEnrollEligible); SingleWorkerOptions localActivityOptions = toLocalActivityOptions( @@ -215,7 +222,8 @@ private static final class TaskSnapshot { contextPropagators, taggedScope, workerInstanceKey, - workerControlTaskQueue); + workerControlTaskQueue, + externalStorage); SlotSupplier workflowSlotSupplier = this.options.getWorkerTuner() == null @@ -915,6 +923,7 @@ private static SingleWorkerOptions toActivityOptions( Scope metricsScope, String workerInstanceKey, String workerControlTaskQueue, + @Nullable ExternalStorage externalStorage, boolean autoEnrollEligible) { return toSingleWorkerOptions( factoryOptions, @@ -922,7 +931,8 @@ private static SingleWorkerOptions toActivityOptions( clientOptions, contextPropagators, workerInstanceKey, - workerControlTaskQueue) + workerControlTaskQueue, + externalStorage) .setUsingVirtualThreads(options.isUsingVirtualThreadsOnActivityWorker()) .setAllowActivityHeartbeatDuringShutdown(options.getAllowActivityHeartbeatDuringShutdown()) .setPollerOptions( @@ -948,6 +958,7 @@ private static SingleWorkerOptions toNexusOptions( Scope metricsScope, String workerInstanceKey, String workerControlTaskQueue, + @Nullable ExternalStorage externalStorage, boolean autoEnrollEligible) { return toSingleWorkerOptions( factoryOptions, @@ -955,7 +966,8 @@ private static SingleWorkerOptions toNexusOptions( clientOptions, contextPropagators, workerInstanceKey, - workerControlTaskQueue) + workerControlTaskQueue, + externalStorage) .setPollerOptions( PollerOptions.newBuilder() .setPollerBehavior( @@ -980,6 +992,7 @@ private static SingleWorkerOptions toWorkflowWorkerOptions( Scope metricsScope, String workerInstanceKey, String workerControlTaskQueue, + @Nullable ExternalStorage externalStorage, boolean autoEnrollEligible) { Map tags = new ImmutableMap.Builder(1).put(MetricsTag.TASK_QUEUE, taskQueue).build(); @@ -1015,7 +1028,8 @@ private static SingleWorkerOptions toWorkflowWorkerOptions( clientOptions, contextPropagators, workerInstanceKey, - workerControlTaskQueue) + workerControlTaskQueue, + externalStorage) .setPollerOptions( PollerOptions.newBuilder() .setPollerBehavior( @@ -1040,14 +1054,16 @@ private static SingleWorkerOptions toLocalActivityOptions( List contextPropagators, Scope metricsScope, String workerInstanceKey, - String workerControlTaskQueue) { + String workerControlTaskQueue, + @Nullable ExternalStorage externalStorage) { return toSingleWorkerOptions( factoryOptions, options, clientOptions, contextPropagators, workerInstanceKey, - workerControlTaskQueue) + workerControlTaskQueue, + externalStorage) .setPollerOptions( PollerOptions.newBuilder() .setPollerBehavior(new PollerBehaviorSimpleMaximum(1)) @@ -1066,7 +1082,8 @@ private static SingleWorkerOptions.Builder toSingleWorkerOptions( WorkflowClientOptions clientOptions, List contextPropagators, String workerInstanceKey, - String workerControlTaskQueue) { + String workerControlTaskQueue, + @Nullable ExternalStorage externalStorage) { String buildId = null; if (options.getBuildId() != null) { buildId = options.getBuildId(); @@ -1081,6 +1098,7 @@ private static SingleWorkerOptions.Builder toSingleWorkerOptions( return SingleWorkerOptions.newBuilder() .setDataConverter(clientOptions.getDataConverter()) + .setExternalStorage(externalStorage) .setIdentity(identity) .setBuildId(buildId) .setUseBuildIdForVersioning(options.isUsingBuildIdForVersioning()) 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/RootWorkflowClientInvokerLinkPropagationTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java index a597ae96d2..93d8d95601 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerLinkPropagationTest.java @@ -66,7 +66,8 @@ public void setUp() { WorkflowClientOptions.newBuilder() .setNamespace(NAMESPACE) .validateAndBuildWithDefaults(), - new WorkerFactoryRegistry()); + new WorkerFactoryRegistry(), + null); Scope metricsScope = new RootScopeBuilder().reportEvery(com.uber.m3.util.Duration.ofMillis(10)); nexusCtx = new InternalNexusOperationContext( diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClientTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClientTest.java new file mode 100644 index 0000000000..e89198d5b3 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ExternalStorageGenericWorkflowClientTest.java @@ -0,0 +1,129 @@ +package io.temporal.internal.client.external; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.grpc.Deadline; +import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.Payloads; +import io.temporal.api.common.v1.WorkflowType; +import io.temporal.api.workflowservice.v1.ExecuteMultiOperationRequest; +import io.temporal.api.workflowservice.v1.ExecuteMultiOperationResponse; +import io.temporal.api.workflowservice.v1.StartActivityExecutionRequest; +import io.temporal.api.workflowservice.v1.StartActivityExecutionResponse; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest; +import io.temporal.internal.payload.storage.ExternalStorage; +import io.temporal.payload.storage.ExternalStorageOptions; +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 java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import org.junit.Test; + +public class ExternalStorageGenericWorkflowClientTest { + + @Test + public void standaloneActivityStartIncludesKnownTargetInfo() { + GenericWorkflowClient next = mock(GenericWorkflowClient.class); + when(next.startActivity(any())).thenReturn(StartActivityExecutionResponse.getDefaultInstance()); + CapturingDriver driver = new CapturingDriver(); + ExternalStorageGenericWorkflowClient client = + new ExternalStorageGenericWorkflowClient( + next, + ExternalStorage.create( + ExternalStorageOptions.newBuilder() + .setDriver(driver) + .setPayloadSizeThreshold(0) + .setMaxConcurrentPayloadVisits(1) + .build()), + "test-namespace"); + StartActivityExecutionRequest request = + StartActivityExecutionRequest.newBuilder() + .setActivityId("activity-id") + .setActivityType(ActivityType.newBuilder().setName("activity-type")) + .setInput(Payloads.newBuilder().addPayloads(Payload.getDefaultInstance())) + .build(); + + client.startActivity(request); + + assertEquals( + Collections.singletonList( + new StorageDriverActivityInfo("test-namespace", "activity-id", null, "activity-type")), + driver.targets); + } + + @Test + public void multiOperationIncludesWorkflowTargetInfo() { + GenericWorkflowClient next = mock(GenericWorkflowClient.class); + when(next.executeMultiOperation(any(), any())) + .thenReturn(ExecuteMultiOperationResponse.getDefaultInstance()); + CapturingDriver driver = new CapturingDriver(); + ExternalStorageGenericWorkflowClient client = + new ExternalStorageGenericWorkflowClient( + next, + ExternalStorage.create( + ExternalStorageOptions.newBuilder() + .setDriver(driver) + .setPayloadSizeThreshold(0) + .setMaxConcurrentPayloadVisits(1) + .build()), + "test-namespace"); + ExecuteMultiOperationRequest request = + ExecuteMultiOperationRequest.newBuilder() + .addOperations( + ExecuteMultiOperationRequest.Operation.newBuilder() + .setStartWorkflow( + StartWorkflowExecutionRequest.newBuilder() + .setWorkflowId("workflow-id") + .setWorkflowType(WorkflowType.newBuilder().setName("workflow-type")) + .setInput( + Payloads.newBuilder().addPayloads(Payload.getDefaultInstance())))) + .build(); + + client.executeMultiOperation(request, Deadline.after(1, TimeUnit.SECONDS)); + + assertEquals( + Collections.singletonList( + new StorageDriverWorkflowInfo("test-namespace", "workflow-id", null, "workflow-type")), + driver.targets); + } + + private static final class CapturingDriver implements StorageDriver { + private final List targets = new ArrayList<>(); + + @Override + public String getName() { + return "test"; + } + + @Override + public String getType() { + return "test"; + } + + @Override + public CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + targets.add(context.getTarget()); + return CompletableFuture.completedFuture( + Collections.singletonList(new StorageDriverClaim(Collections.emptyMap()))); + } + + @Override + public CompletableFuture> retrieve( + StorageDriverRetrieveContext context, List claims) { + return CompletableFuture.completedFuture(Collections.emptyList()); + } + } +} 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..1c190f1463 --- /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.ExternalStorage; +import io.temporal.payload.storage.ExternalStorageOptions; +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 ExternalStorage externalStorage; + + @Before + public void setUp() { + service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.getDefaultInstance()); + when(service.getOptions()).thenReturn(WorkflowServiceStubsOptions.getDefaultInstance()); + externalStorage = + ExternalStorage.create( + ExternalStorageOptions.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/worker/ActivityWorkerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerTest.java new file mode 100644 index 0000000000..0166586fd3 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/ActivityWorkerTest.java @@ -0,0 +1,45 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.assertEquals; + +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 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); + } +} 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..3e87683187 100644 --- a/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java +++ b/temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java @@ -30,6 +30,7 @@ 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.sync.*; import io.temporal.internal.testservice.InProcessGRPCServer; @@ -100,16 +101,19 @@ public TestActivityEnvironmentInternal(@Nullable TestEnvironmentOptions options) this.workflowServiceStubs = WorkflowServiceStubs.newServiceStubs(serviceStubsOptionsBuilder.build()); + WorkflowClient client = + WorkflowClient.newInstance( + this.workflowServiceStubs, testEnvironmentOptions.getWorkflowClientOptions()); 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); + heartbeatExecutor, + ((WorkflowClientInternal) client.getInternal()).getExternalStorage()); activityTaskHandler = new ActivityTaskHandlerImpl( testEnvironmentOptions.getWorkflowClientOptions().getNamespace(),