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..6f0b72f63d 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionDescription.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionDescription.java @@ -1,8 +1,8 @@ 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; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -29,15 +29,8 @@ public String getStaticSummary() { if (!response.getExecutionConfig().getUserMetadata().hasSummary()) { return null; } - return dataConverter - .withContext( - new WorkflowSerializationContext( - response.getWorkflowExecutionInfo().getParentNamespaceId(), - response.getWorkflowExecutionInfo().getExecution().getWorkflowId())) - .fromPayload( - response.getExecutionConfig().getUserMetadata().getSummary(), - String.class, - String.class); + Payload summary = response.getExecutionConfig().getUserMetadata().getSummary(); + return dataConverter.fromPayload(summary, String.class, String.class); } /** @@ -51,15 +44,8 @@ public String getStaticDetails() { if (!response.getExecutionConfig().getUserMetadata().hasDetails()) { return null; } - return dataConverter - .withContext( - new WorkflowSerializationContext( - response.getWorkflowExecutionInfo().getParentNamespaceId(), - response.getWorkflowExecutionInfo().getExecution().getWorkflowId())) - .fromPayload( - response.getExecutionConfig().getUserMetadata().getDetails(), - String.class, - String.class); + Payload details = response.getExecutionConfig().getUserMetadata().getDetails(); + return dataConverter.fromPayload(details, String.class, String.class); } /** Returns the raw response from the Temporal service. */ diff --git a/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionMetadata.java b/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionMetadata.java index b35cdaed12..1fd96a0497 100644 --- a/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionMetadata.java +++ b/temporal-sdk/src/main/java/io/temporal/client/WorkflowExecutionMetadata.java @@ -9,7 +9,6 @@ import io.temporal.common.converter.DataConverter; import io.temporal.internal.common.ProtobufTimeUtils; import io.temporal.internal.common.SearchAttributesUtil; -import io.temporal.payload.context.WorkflowSerializationContext; import java.lang.reflect.Type; import java.time.Duration; import java.time.Instant; @@ -123,11 +122,7 @@ public T getMemo(String key, Class valueClass, Type genericType) { if (memo == null) { return null; } - return dataConverter - .withContext( - new WorkflowSerializationContext( - info.getParentNamespaceId(), info.getExecution().getWorkflowId())) - .fromPayload(memo, valueClass, genericType); + return dataConverter.fromPayload(memo, valueClass, genericType); } @Nonnull 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..c41f3740d9 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,8 @@ package io.temporal.internal.activity; +import com.google.common.base.Strings; +import com.google.common.base.Throwables; +import com.google.protobuf.ByteString; import com.uber.m3.tally.Scope; import io.grpc.Status; import io.grpc.StatusRuntimeException; @@ -7,6 +10,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,22 +18,38 @@ 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.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; 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; @ThreadSafe class HeartbeatContextImpl implements HeartbeatContext { + private static final class HeartbeatAbandonedException extends RuntimeException { + HeartbeatAbandonedException() { + super(null, null, false, false); + } + } + private static final Logger log = LoggerFactory.getLogger(HeartbeatContextImpl.class); private static final long HEARTBEAT_RETRY_WAIT_MILLIS = 1000; // Buffer added to the heartbeat timeout to avoid racing with the server's own timeout tracking. @@ -58,6 +78,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; @@ -76,6 +97,9 @@ static long getLocalHeartbeatTimeoutBufferMillis() { private boolean heartbeatTimedOut; private boolean rejectNewHeartbeats; + private volatile CompletableFuture outstandingOffloadAbandon; + private final AtomicInteger pendingAbandons = new AtomicInteger(); + private ActivityCompletionException lastException; private final CancelSource cancellationSource = new CancelSource<>(ActivityCanceledException::new); @@ -89,7 +113,8 @@ public HeartbeatContextImpl( Scope metricsScope, String identity, Duration maxHeartbeatThrottleInterval, - Duration defaultHeartbeatThrottleInterval) { + Duration defaultHeartbeatThrottleInterval, + @Nullable ExternalStorageRunner externalStorage) { this( service, namespace, @@ -100,6 +125,7 @@ public HeartbeatContextImpl( identity, maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, + externalStorage, getLocalHeartbeatTimeoutBufferMillis()); } @@ -113,10 +139,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( @@ -151,7 +179,9 @@ public void heartbeat(V details) throws ActivityCompletionException { if (heartbeatExecutor.isShutdown()) { throw new ActivityWorkerShutdownException(info); } + requestOffloadAbandon(); lock.lock(); + pendingAbandons.decrementAndGet(); try { checkHeartbeatTimeoutDeadlineLocked(); if (rejectNewHeartbeats) { @@ -224,7 +254,9 @@ public Object getLatestHeartbeatDetails() { @Override public void cancelOutstandingHeartbeat() { + requestOffloadAbandon(); lock.lock(); + pendingAbandons.decrementAndGet(); try { if (scheduledHeartbeat != null) { scheduledHeartbeat.cancel(false); @@ -239,7 +271,9 @@ public void cancelOutstandingHeartbeat() { @Override public void cancelFromWorkerCommand() { + requestOffloadAbandon(); lock.lock(); + pendingAbandons.decrementAndGet(); try { requestCancelLocked(); } finally { @@ -274,6 +308,9 @@ private void doHeartBeatLocked(Object details) { if (heartbeatTimeoutDeadlineNanos != 0) { heartbeatTimeoutDeadlineNanos = computeHeartbeatTimeoutDeadlineNanos(); } + } catch (HeartbeatAbandonedException e) { + scheduledHeartbeat = null; + return; } catch (StatusRuntimeException e) { // Not rethrowing to not fail activity implementation on intermittent connection or Temporal // errors. @@ -330,16 +367,97 @@ 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, if a newer heartbeat supersedes this one, or if the activity is cancelled. + */ + private void offloadHeartbeat(RecordActivityTaskHeartbeatRequest.Builder builder) { + CancelSource offloadCancel = + new CancelSource<>(CancellationException::new); + CompletableFuture abandon = new CompletableFuture<>(); + CancellationToken.Registration onActivityCancel = + cancellationSource + .token() + .onCancel( + () -> { + offloadCancel.cancel(); + abandon.complete(null); + }); + outstandingOffloadAbandon = abandon; + try { + if (pendingAbandons.get() > 0) { + throw new HeartbeatAbandonedException(); + } + CompletableFuture store = + externalStorage.storeAsync(builder, activityStorageTarget(), null, offloadCancel.token()); + CompletableFuture.anyOf(store, abandon).get(heartbeatIntervalMillis, TimeUnit.MILLISECONDS); + if (!store.isDone()) { + offloadCancel.cancel(); + throw new HeartbeatAbandonedException(); + } + store.get(); + } catch (TimeoutException e) { + offloadCancel.cancel(); + throw new CancellationException( + "External storage did not store the heartbeat details within the heartbeat interval"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + offloadCancel.cancel(); + CancellationException cancelled = + new CancellationException("External storage store interrupted"); + cancelled.initCause(e); + throw cancelled; + } catch (ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + Throwables.throwIfUnchecked(cause); + throw new CompletionException(cause); + } finally { + outstandingOffloadAbandon = null; + onActivityCancel.close(); + } + } + + private void requestOffloadAbandon() { + pendingAbandons.incrementAndGet(); + CompletableFuture outstanding = outstandingOffloadAbandon; + if (outstanding != null) { + outstanding.complete(null); + } + } + 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..129e1c2df2 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 @@ -5,11 +5,14 @@ import static io.temporal.internal.common.HeaderUtils.intoPayloadMap; import static io.temporal.internal.common.WorkflowExecutionUtils.makeUserMetaData; +import com.google.common.base.Strings; import com.google.common.collect.Iterators; import io.grpc.Deadline; 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; @@ -19,6 +22,7 @@ import io.temporal.api.update.v1.*; import io.temporal.api.workflowservice.v1.*; import io.temporal.client.*; +import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.common.interceptors.WorkflowClientCallsInterceptor; import io.temporal.internal.client.external.GenericWorkflowClient; @@ -28,8 +32,9 @@ 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.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.StatusUtils; import io.temporal.worker.WorkflowTaskDispatchHandle; import java.lang.reflect.Type; @@ -47,28 +52,73 @@ 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; + private final @Nullable ExternalStorageRunner externalStorage; 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.externalStorage = 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); + } + + private void storeHeader( + Header.Builder header, + String workflowId, + @Nullable String runId, + @Nullable String workflowType) { + if (externalStorage == null || header.getFieldsCount() == 0) { + return; + } + externalStorage.store( + header, + new StorageDriverWorkflowInfo( + namespace, + Strings.emptyToNull(workflowId), + Strings.emptyToNull(runId), + Strings.emptyToNull(workflowType)), + null, + CancellationToken.none()); + } + @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 +176,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,15 +188,15 @@ 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); + storeHeader( + request.getHeaderBuilder(), + input.getWorkflowExecution().getWorkflowId(), + input.getWorkflowExecution().getRunId(), + null); SignalWorkflowExecutionResponse response = genericClient.signal(request.build()); // Server >=1.31 with EnableCHASMSignalBacklinks returns a response link pointing at the signal // event; older servers leave it unset. Propagate when present. @@ -161,11 +211,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 +251,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() @@ -342,24 +385,27 @@ private StartWorkflowExecutionRequest.Builder toStartRequest( workflowStartInput.getOptions().getStaticDetails(), dataConverterWithWorkflowContext); - return requestsHelper.newStartWorkflowExecutionRequest( + StartWorkflowExecutionRequest.Builder startRequest = + requestsHelper.newStartWorkflowExecutionRequest( + workflowStartInput.getWorkflowId(), + workflowStartInput.getWorkflowType(), + workflowStartInput.getHeader(), + workflowStartInput.getOptions(), + workflowInput.orElse(null), + memo, + userMetadata); + storeHeader( + startRequest.getHeaderBuilder(), workflowStartInput.getWorkflowId(), - workflowStartInput.getWorkflowType(), - workflowStartInput.getHeader(), - workflowStartInput.getOptions(), - workflowInput.orElse(null), - memo, - userMetadata); + null, + workflowStartInput.getWorkflowType()); + return startRequest; } @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 +426,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 +453,25 @@ 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); + storeHeader( + query.getHeaderBuilder(), + input.getWorkflowExecution().getWorkflowId(), + input.getWorkflowExecution().getRunId(), + null); 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 +494,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); @@ -511,13 +550,15 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null)) .setName(input.getUpdateName()); inputArgs.ifPresent(updateInput::setArgs); + storeHeader( + updateInput.getHeaderBuilder(), + input.getWorkflowExecution().getWorkflowId(), + input.getWorkflowExecution().getRunId(), + null); 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 +571,7 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( try { nexusOperationMetadata.operationToken = OperationTokenUtil.generateWorkflowUpdateOperationToken( - clientOptions.getNamespace(), + namespace, input.getWorkflowExecution().getWorkflowId(), input.getWorkflowExecution().getRunId(), input.getUpdateId()); @@ -555,7 +596,7 @@ private UpdateWorkflowExecutionRequest toUpdateWorkflowExecutionRequest( Request request = requestBuilder.build(); return UpdateWorkflowExecutionRequest.newBuilder() - .setNamespace(clientOptions.getNamespace()) + .setNamespace(namespace) .setWaitPolicy(input.getWaitPolicy()) .setWorkflowExecution( WorkflowExecution.newBuilder() @@ -622,11 +663,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 +680,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 +757,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 +773,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 +783,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 +795,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 +811,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 +823,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..9384be6e00 --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/WorkflowClientDataConverterFactory.java @@ -0,0 +1,37 @@ +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; + + WorkflowClientDataConverterFactory( + WorkflowClientOptions clientOptions, @Nullable ExternalStorageRunner externalStorage) { + this.namespace = clientOptions.getNamespace(); + this.baseConverter = + new ExternalStorageDataConverter(clientOptions.getDataConverter(), externalStorage); + } + + DataConverter forWorkflow( + String workflowId, @Nullable String runId, @Nullable String workflowType) { + DataConverter converter = + baseConverter.withContext(new WorkflowSerializationContext(namespace, workflowId)); + 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..5e597cbb58 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); + RecordActivityTaskHeartbeatRequest request = storeOutbound(builder.build()); RecordActivityTaskHeartbeatResponse status = - ActivityClientHelper.sendHeartbeatRequest( - service, - namespace, - identity, - taskToken, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope); + ActivityClientHelper.sendHeartbeatRequest(service, request, 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); + RecordActivityTaskHeartbeatByIdRequest request = storeOutbound(builder.build()); RecordActivityTaskHeartbeatByIdResponse status = - ActivityClientHelper.recordActivityTaskHeartbeatById( - service, - namespace, - identity, - execution, - activityId, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope); + ActivityClientHelper.recordActivityTaskHeartbeatById(service, request, 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/payload/storage/ExternalStorageDataConverter.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java index 4c725752a2..ab1c616179 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageDataConverter.java @@ -22,17 +22,17 @@ public final class ExternalStorageDataConverter implements DataConverter { private final DataConverter delegate; - private final ExternalStorageRunner externalStorage; + private final @Nullable ExternalStorageRunner externalStorage; private final @Nullable StorageDriverTargetInfo storageTarget; public ExternalStorageDataConverter( - @Nonnull DataConverter delegate, @Nonnull ExternalStorageRunner externalStorage) { + @Nonnull DataConverter delegate, @Nullable ExternalStorageRunner externalStorage) { this(delegate, externalStorage, null); } private ExternalStorageDataConverter( @Nonnull DataConverter delegate, - @Nonnull ExternalStorageRunner externalStorage, + @Nullable ExternalStorageRunner externalStorage, @Nullable StorageDriverTargetInfo storageTarget) { this.delegate = delegate; this.externalStorage = externalStorage; @@ -127,16 +127,25 @@ private Payload retrieve(Payload payload) { } private Payloads store(Payloads payloads) { + if (externalStorage == null) { + return payloads; + } Payloads.Builder builder = payloads.toBuilder(); externalStorage.store(builder, storageTarget, null, CancellationToken.none()); return builder.build(); } private T retrieveMessage(T message) { + if (externalStorage == null) { + throw new ExternalStorageNotConfiguredException(); + } return externalStorage.retrieve(message, CancellationToken.none()); } private Failure storeMessage(Failure failure) { + if (externalStorage == null) { + return failure; + } Failure.Builder builder = failure.toBuilder(); externalStorage.store(builder, storageTarget, null, CancellationToken.none()); return builder.build(); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java index c05f8c5d03..8fbe763cfa 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageRunner.java @@ -45,8 +45,15 @@ public void store( @Nullable MessageVisitor targetVisitor, CancellationToken cancellationToken) { getOrThrowIfCancelled( - PayloadVisitors.visit(builder, storeOptions(target, targetVisitor, cancellationToken)), - cancellationToken); + storeAsync(builder, target, targetVisitor, cancellationToken), cancellationToken); + } + + public CompletableFuture storeAsync( + Message.Builder builder, + @Nullable StorageDriverTargetInfo target, + @Nullable MessageVisitor targetVisitor, + CancellationToken cancellationToken) { + return PayloadVisitors.visit(builder, storeOptions(target, targetVisitor, cancellationToken)); } public T retrieve( 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..db71a18bf4 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,8 +367,19 @@ public void handle(ActivityTask task) throws Exception { } private ActivityTaskHandler.Result handleActivity(ActivityTask task, Scope metricsScope) { + ByteString taskToken = task.getResponse().getTaskToken(); + try { + task = retrieveInboundPayloads(task); + } catch (Exception e) { + RespondActivityTaskFailedRequest sent = sendStorageFailure(taskToken, metricsScope, e); + return new ActivityTaskHandler.Result( + task.getResponse().getActivityId(), + null, + new ActivityTaskHandler.Result.TaskFailedResult(sent, e), + null, + false); + } PollActivityTaskQueueResponseOrBuilder pollResponse = task.getResponse(); - ByteString taskToken = pollResponse.getTaskToken(); metricsScope .timer(MetricsType.ACTIVITY_SCHEDULE_TO_START_LATENCY) .record( @@ -354,7 +401,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, 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 +442,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 +467,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 +486,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 +508,73 @@ 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 RespondActivityTaskFailedRequest sendStorageFailure( + ByteString taskToken, 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 request = + RespondActivityTaskFailedRequest.newBuilder() + .setTaskToken(taskToken) + .setIdentity(options.getIdentity()) + .setNamespace(namespace) + .setWorkerVersion(options.workerVersionStamp()) + .setFailure(options.getDataConverter().exceptionToFailure(applicationFailure)) + .build(); + grpcRetryer.retry( + () -> + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .respondActivityTaskFailed(request), + replyGrpcRetryerOptions); + return request; + } + + @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..1843876fc6 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/client/WorkflowExecutionMetadataTest.java @@ -0,0 +1,190 @@ +package io.temporal.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +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.context.SerializationContext; +import io.temporal.payload.context.WorkflowSerializationContext; +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 java.util.concurrent.atomic.AtomicReference; +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)); + } + + @Test + public void getMemoDecodesWithTheContextSuppliedByTheCaller() { + AtomicReference seen = new AtomicReference<>(); + DataConverter base = DefaultDataConverter.newDefaultInstance(); + Payload inline = base.toPayloads("plain").get().getPayloads(0); + WorkflowExecutionInfo info = + WorkflowExecutionInfo.newBuilder() + .setMemo(Memo.newBuilder().putFields("k", inline)) + .build(); + + DataConverter contextual = + new ContextRecordingDataConverter(base, null, seen) + .withContext(new WorkflowSerializationContext("the-namespace", "wf-1")); + + assertEquals( + "plain", new WorkflowExecutionMetadata(info, contextual).getMemo("k", String.class)); + + SerializationContext used = seen.get(); + assertNotNull("the converter should have been used with a context", used); + assertEquals( + "the caller's namespace must survive to the codec", + "the-namespace", + ((WorkflowSerializationContext) used).getNamespace()); + assertEquals("wf-1", ((WorkflowSerializationContext) used).getWorkflowId()); + } + + 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); + } + } + + private static final class ContextRecordingDataConverter implements DataConverter { + private final DataConverter delegate; + private final SerializationContext context; + private final AtomicReference seen; + + ContextRecordingDataConverter( + DataConverter delegate, + SerializationContext context, + AtomicReference seen) { + this.delegate = delegate; + this.context = context; + this.seen = seen; + } + + @Override + public DataConverter withContext(SerializationContext context) { + return new ContextRecordingDataConverter(delegate, context, seen); + } + + @Override + public java.util.Optional toPayload(T value) { + return delegate.toPayload(value); + } + + @Override + public T fromPayload( + Payload payload, Class valueClass, java.lang.reflect.Type valueType) { + seen.set(context); + return delegate.fromPayload(payload, valueClass, valueType); + } + + @Override + public java.util.Optional toPayloads(Object... values) { + return delegate.toPayloads(values); + } + + @Override + public T fromPayloads( + int index, + java.util.Optional content, + Class parameterType, + java.lang.reflect.Type genericParameterType) { + seen.set(context); + return delegate.fromPayloads(index, content, parameterType, genericParameterType); + } + + @Override + public io.temporal.api.failure.v1.Failure exceptionToFailure(Throwable throwable) { + return delegate.exceptionToFailure(throwable); + } + + @Override + public RuntimeException failureToException(io.temporal.api.failure.v1.Failure failure) { + return delegate.failureToException(failure); + } + } +} 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..19fac4d71a 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 @@ -8,6 +8,7 @@ import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.temporal.activity.ActivityInfo; +import io.temporal.api.common.v1.Payload; import io.temporal.api.enums.v1.TimeoutType; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatRequest; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; @@ -18,14 +19,29 @@ import io.temporal.common.CancellationToken; import io.temporal.common.converter.GlobalDataConverter; import io.temporal.failure.TimeoutFailure; +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.payload.storage.StorageDriverWorkflowInfo; import io.temporal.serviceclient.WorkflowServiceStubs; import io.temporal.testUtils.Eventually; 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.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.After; import org.junit.Before; @@ -329,7 +345,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 +380,7 @@ private HeartbeatContextImpl createHeartbeatContext( "test-identity", maxHeartbeatThrottleInterval, defaultHeartbeatThrottleInterval, + null, TEST_BUFFER_MILLIS); } @@ -390,4 +408,168 @@ 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)); + } + + private static final Duration OFFLOAD_INTERVAL = Duration.ofMillis(500); + + @Test + public void aNewerHeartbeatAbandonsAnInFlightOffload() throws Exception { + BlockingDriver driver = new BlockingDriver(); + HeartbeatContextImpl ctx = createHeartbeatContextWithStorage(driver); + + Thread first = new Thread(() -> ctx.heartbeat(largeDetails("first"))); + first.start(); + assertTrue("the first store should start", driver.storeStarted.await(5, TimeUnit.SECONDS)); + + ctx.heartbeat(largeDetails("second")); + first.join(5000); + assertFalse("the superseded heartbeat should not still be running", first.isAlive()); + + assertEquals("both heartbeats should attempt a store", 2, driver.stores.get()); + verify(blockingStub, times(1)) + .recordActivityTaskHeartbeat(any(RecordActivityTaskHeartbeatRequest.class)); + assertTrue( + "the abandoned store should have its cancellation token tripped", driver.firstCancelled()); + } + + @Test + public void aThrottledOffloadIsAbandonedWhenTheHeartbeatPoolHasNoSpareThread() throws Exception { + when(blockingStub.recordActivityTaskHeartbeat(any(RecordActivityTaskHeartbeatRequest.class))) + .thenReturn(RecordActivityTaskHeartbeatResponse.getDefaultInstance()); + BlockingDriver driver = new BlockingDriver(); + HeartbeatContextImpl ctx = createHeartbeatContextWithStorage(driver); + + ctx.heartbeat("small"); + ctx.heartbeat(largeDetails("throttled")); + + assertTrue( + "the throttled heartbeat should offload on the heartbeat pool", + driver.storeStarted.await(5, TimeUnit.SECONDS)); + Eventually.assertEventually( + Duration.ofSeconds(5), + () -> + assertTrue( + "the offload must be abandoned even though it occupies the only pool thread", + driver.firstCancelled())); + } + + @Test + public void activityCancellationAbandonsAnInFlightOffload() throws Exception { + BlockingDriver driver = new BlockingDriver(); + HeartbeatContextImpl ctx = createHeartbeatContextWithStorage(driver); + + Thread heartbeating = new Thread(() -> ctx.heartbeat(largeDetails("cancelled"))); + heartbeating.start(); + assertTrue("the store should start", driver.storeStarted.await(5, TimeUnit.SECONDS)); + + long startNanos = System.nanoTime(); + ctx.cancelFromWorkerCommand(); + heartbeating.join(5000); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + assertFalse("cancellation should not wait for the driver", heartbeating.isAlive()); + assertTrue( + "cancellation should abandon the store rather than wait out the heartbeat interval, took " + + elapsedMillis + + "ms", + elapsedMillis < OFFLOAD_INTERVAL.toMillis() / 2); + assertTrue("the abandoned store should be cancelled", driver.firstCancelled()); + } + + private HeartbeatContextImpl createHeartbeatContextWithStorage(StorageDriver driver) { + ExternalStorageRunner runner = + ExternalStorageRunner.create( + ExternalStorage.newBuilder().setDriver(driver).setPayloadSizeThreshold(100).build()); + return new HeartbeatContextImpl( + service, + "test-namespace", + activityInfoWithHeartbeatTimeout(Duration.ZERO), + GlobalDataConverter.get(), + heartbeatExecutor, + new NoopScope(), + "test-identity", + Duration.ofSeconds(60), + OFFLOAD_INTERVAL, + runner, + TEST_BUFFER_MILLIS); + } + + private static String largeDetails(String tag) { + return tag + String.join("", Collections.nCopies(20, "0123456789")); + } + + /** Never completes its first store, so the SDK has to abandon it. */ + private static final class BlockingDriver implements StorageDriver { + final CountDownLatch storeStarted = new CountDownLatch(1); + final AtomicInteger stores = new AtomicInteger(); + private final Map objects = new HashMap<>(); + private volatile CancellationToken firstStoreToken; + private int counter = 0; + + boolean firstCancelled() { + CancellationToken token = firstStoreToken; + return token != null && token.isCancellationRequested(); + } + + @Override + public String getName() { + return "blocking"; + } + + @Override + public String getType() { + return "test.blocking"; + } + + @Override + public synchronized CompletableFuture> store( + StorageDriverStoreContext context, List payloads) { + if (stores.getAndIncrement() == 0) { + firstStoreToken = context.getCancellationToken(); + storeStarted.countDown(); + return new CompletableFuture<>(); + } + 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/client/RootWorkflowClientInvokerStorageTargetTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java new file mode 100644 index 0000000000..488c573349 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootWorkflowClientInvokerStorageTargetTest.java @@ -0,0 +1,265 @@ +package io.temporal.internal.client; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.temporal.api.common.v1.Payload; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.QueryWorkflowRequest; +import io.temporal.api.workflowservice.v1.QueryWorkflowResponse; +import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse; +import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest; +import io.temporal.api.workflowservice.v1.StartWorkflowExecutionRequest; +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.QueryInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalInput; +import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.WorkflowSignalWithStartInput; +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; +import org.mockito.ArgumentCaptor; + +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()); + } + + @Test + public void signalOffloadsHeaderPayloads() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + ArgumentCaptor sent = + ArgumentCaptor.forClass(SignalWorkflowExecutionRequest.class); + + invoker(rpc, driver) + .signal( + new WorkflowSignalInput( + WorkflowExecution.newBuilder().setWorkflowId("wf-h").setRunId("run-h").build(), + "mySignal", + headerWith("trace", "some-tracing-context"), + new Object[] {"argument"})); + + verify(rpc).signal(sent.capture()); + Payload original = tracePayload("some-tracing-context"); + assertTrue("the header value must reach the driver", driver.stored.contains(original)); + assertNotEquals( + "the sent header must be a reference, not the original bytes", + original, + sent.getValue().getHeader().getFieldsOrThrow("trace")); + } + + @Test + public void queryOffloadsHeaderPayloads() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + when(rpc.query(any())).thenReturn(QueryWorkflowResponse.getDefaultInstance()); + ArgumentCaptor sent = ArgumentCaptor.forClass(QueryWorkflowRequest.class); + + invoker(rpc, driver) + .query( + new QueryInput<>( + WorkflowExecution.newBuilder().setWorkflowId("wf-q").build(), + "myQuery", + headerWith("trace", "some-tracing-context"), + new Object[] {"argument"}, + String.class, + String.class)); + + verify(rpc).query(sent.capture()); + Payload original = tracePayload("some-tracing-context"); + assertTrue("the header value must reach the driver", driver.stored.contains(original)); + assertNotEquals( + "the sent header must be a reference, not the original bytes", + original, + sent.getValue().getQuery().getHeader().getFieldsOrThrow("trace")); + } + + @Test + public void startOffloadsHeaderPayloads() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + when(rpc.start(any())).thenReturn(StartWorkflowExecutionResponse.getDefaultInstance()); + ArgumentCaptor sent = + ArgumentCaptor.forClass(StartWorkflowExecutionRequest.class); + + invoker(rpc, driver) + .start( + new WorkflowStartInput( + "wf-s", + "MyWorkflowType", + headerWith("trace", "some-tracing-context"), + new Object[] {"argument"}, + WorkflowOptions.newBuilder().setTaskQueue("tq").build())); + + verify(rpc).start(sent.capture()); + Payload original = tracePayload("some-tracing-context"); + assertTrue("the header value must reach the driver", driver.stored.contains(original)); + assertNotEquals( + "a start header lands in history, so it must be offloaded", + original, + sent.getValue().getHeader().getFieldsOrThrow("trace")); + } + + @Test + public void signalWithStartOffloadsHeaderPayloads() { + CapturingDriver driver = new CapturingDriver(); + GenericWorkflowClient rpc = mock(GenericWorkflowClient.class); + when(rpc.signalWithStart(any())) + .thenReturn(SignalWithStartWorkflowExecutionResponse.getDefaultInstance()); + ArgumentCaptor sent = + ArgumentCaptor.forClass(SignalWithStartWorkflowExecutionRequest.class); + + invoker(rpc, driver) + .signalWithStart( + new WorkflowSignalWithStartInput( + new WorkflowStartInput( + "wf-sws", + "MyWorkflowType", + headerWith("trace", "some-tracing-context"), + new Object[] {"argument"}, + WorkflowOptions.newBuilder().setTaskQueue("tq").build()), + "mySignal", + new Object[] {"signal-arg"})); + + verify(rpc).signalWithStart(sent.capture()); + Payload original = tracePayload("some-tracing-context"); + assertTrue("the header value must reach the driver", driver.stored.contains(original)); + assertNotEquals( + "the copied start header must carry the reference", + original, + sent.getValue().getHeader().getFieldsOrThrow("trace")); + } + + private static Header headerWith(String key, String value) { + return new Header(Collections.singletonMap(key, tracePayload(value))); + } + + private static Payload tracePayload(String value) { + return WorkflowClientOptions.newBuilder() + .validateAndBuildWithDefaults() + .getDataConverter() + .toPayload(value) + .get(); + } + + 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; + final List stored = Collections.synchronizedList(new ArrayList<>()); + 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(); + stored.addAll(payloads); + 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/payload/storage/ExternalStorageDataConverterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageDataConverterTest.java index b4fd9cc787..e0abbe32ca 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageDataConverterTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageDataConverterTest.java @@ -4,6 +4,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import com.google.protobuf.ByteString; @@ -212,6 +213,46 @@ private static ExternalStorageRunner runner(StorageDriver driver, int threshold) } /** Obscures payload bytes so plaintext reaching a driver is detectable. */ + @Test + public void withoutStorageAReferenceRaisesTheNotConfiguredError() { + Payload reference = storeAndTakeReference("offloaded"); + DataConverter unconfigured = new ExternalStorageDataConverter(plain, null); + + ExternalStorageNotConfiguredException thrown = + assertThrows( + ExternalStorageNotConfiguredException.class, + () -> unconfigured.fromPayload(reference, String.class, String.class)); + + assertTrue( + "the error should point at the option that fixes it", + thrown.getMessage().contains("TMPRL1105") + && thrown.getMessage().contains("setExternalStorage")); + } + + @Test + public void withoutStorageInlinePayloadsStillRoundTrip() { + DataConverter unconfigured = new ExternalStorageDataConverter(plain, null); + + Payload inline = unconfigured.toPayload("plain").get(); + + assertEquals("plain", unconfigured.fromPayload(inline, String.class, String.class)); + assertNull( + "nothing should be offloaded when storage is not configured", + ExternalStorageReferences.tryParseReference(inline)); + } + + private Payload storeAndTakeReference(String value) { + ExternalStorageDataConverter configured = + new ExternalStorageDataConverter( + plain, + ExternalStorageRunner.create( + ExternalStorage.newBuilder() + .setDriver(new RecordingDriver()) + .setPayloadSizeThreshold(0) + .build())); + return configured.toPayload(value).get(); + } + private static final class CountingCodec implements PayloadCodec { private static final byte KEY = 0x5A; diff --git a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java index 7cef800eb5..891c4de40f 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/payload/storage/ExternalStorageRunnerTest.java @@ -13,10 +13,12 @@ import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributesOrBuilder; import io.temporal.api.command.v1.StartChildWorkflowExecutionCommandAttributes; import io.temporal.api.common.v1.ActivityType; +import io.temporal.api.common.v1.Header; import io.temporal.api.common.v1.Payload; import io.temporal.api.common.v1.Payloads; import io.temporal.api.common.v1.SearchAttributes; import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest; +import io.temporal.api.workflowservice.v1.SignalWorkflowExecutionRequest; import io.temporal.common.CancellationToken; import io.temporal.internal.concurrent.structured.CancelSource; import io.temporal.internal.payload.visitor.MessageVisitor; @@ -59,6 +61,49 @@ public void storeAndRetrieveRoundTripsOverAMessage() throws Exception { assertEquals(message, retrieved); } + @Test + public void storeOffloadsHeaders() throws Exception { + InMemoryDriver driver = new InMemoryDriver("d1"); + ExternalStorageRunner transformer = transformer(driver, 0); + SignalWorkflowExecutionRequest request = + SignalWorkflowExecutionRequest.newBuilder() + .setHeader(Header.newBuilder().putFields("trace", payload("ctx"))) + .setInput(Payloads.newBuilder().addPayloads(payload("arg"))) + .build(); + + SignalWorkflowExecutionRequest.Builder builder = request.toBuilder(); + transformer.store(builder, null, null, CancellationToken.none()); + SignalWorkflowExecutionRequest stored = builder.build(); + + assertNotNull( + "headers must be offloaded like any other payload", + ExternalStorageReferences.tryParseReference(stored.getHeader().getFieldsOrThrow("trace"))); + assertNotNull( + "input must be offloaded", + ExternalStorageReferences.tryParseReference(stored.getInput().getPayloads(0))); + } + + @Test + public void retrieveStillResolvesAHeaderStoredElsewhere() throws Exception { + InMemoryDriver driver = new InMemoryDriver("d1"); + ExternalStorageRunner transformer = transformer(driver, 0); + + Payloads.Builder headerValue = Payloads.newBuilder().addPayloads(payload("ctx")); + transformer.store(headerValue, null, null, CancellationToken.none()); + Payload storedHeader = headerValue.build().getPayloads(0); + assertNotNull(ExternalStorageReferences.tryParseReference(storedHeader)); + + SignalWorkflowExecutionRequest request = + SignalWorkflowExecutionRequest.newBuilder() + .setHeader(Header.newBuilder().putFields("trace", storedHeader)) + .build(); + + SignalWorkflowExecutionRequest retrieved = + transformer.retrieve(request, CancellationToken.none()); + + assertEquals(payload("ctx"), retrieved.getHeader().getFieldsOrThrow("trace")); + } + @Test public void walksNestedPayloads() throws Exception { InMemoryDriver driver = new InMemoryDriver("d1"); 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(