exstore/pipeline integration

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientInternalImpl.java

@@ -18,12 +18,15 @@
 import io.temporal.internal.WorkflowThreadMarker;
 import io.temporal.internal.client.*;
 import io.temporal.internal.client.NexusStartWorkflowResponse;
+import io.temporal.internal.client.external.ExternalStorageGenericWorkflowClient;
 import io.temporal.internal.client.external.GenericWorkflowClient;
 import io.temporal.internal.client.external.GenericWorkflowClientImpl;
 import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
 import io.temporal.internal.common.PluginUtils;
+import io.temporal.internal.payload.storage.ExternalStorageMessageConverter;
 import io.temporal.internal.sync.StubMarker;
 import io.temporal.internal.worker.HeartbeatManager;
+import io.temporal.payload.storage.ExternalStorageOptions;
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.serviceclient.WorkflowServiceStubsPlugin;
@@ -106,15 +109,26 @@ public static WorkflowClient newInstance(
             .getOptions()
             .getMetricsScope()
             .tagged(MetricsTag.defaultTags(options.getNamespace()));
-    this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
+    ExternalStorageOptions externalStorage = options.getExternalStorage();
+    ExternalStorageMessageConverter externalStorageConverter =
+        externalStorage == null ? null : ExternalStorageMessageConverter.create(externalStorage);
+    GenericWorkflowClient genericClient =
+        new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
+    if (externalStorageConverter != null) {
+      genericClient =
+          new ExternalStorageGenericWorkflowClient(
+              genericClient, externalStorageConverter, options.getNamespace());
+    }
+    this.genericClient = genericClient;
     this.interceptors = options.getInterceptors();
     this.workflowClientCallsInvoker = initializeClientInvoker();
     this.manualActivityCompletionClientFactory =
         ManualActivityCompletionClientFactory.newFactory(
             workflowServiceStubs,
             options.getNamespace(),
             options.getIdentity(),
-            options.getDataConverter());
+            options.getDataConverter(),
+            externalStorageConverter);
 
     java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval();
     if (!heartbeatInterval.isNegative()) {

temporal-sdk/src/main/java/io/temporal/client/WorkflowClientOptions.java

@@ -7,6 +7,7 @@
 import io.temporal.common.converter.DataConverter;
 import io.temporal.common.converter.GlobalDataConverter;
 import io.temporal.common.interceptors.WorkflowClientInterceptor;
+import io.temporal.payload.storage.ExternalStorageOptions;
 import java.lang.management.ManagementFactory;
 import java.time.Duration;
 import java.util.Arrays;
@@ -52,6 +53,7 @@ public static final class Builder {
     private QueryRejectCondition queryRejectCondition;
     private WorkflowClientPlugin[] plugins;
     private Duration workerHeartbeatInterval;
+    private ExternalStorageOptions externalStorage;
 
     private Builder() {}
 
@@ -68,6 +70,7 @@ private Builder(WorkflowClientOptions options) {
       queryRejectCondition = options.queryRejectCondition;
       plugins = options.plugins;
       workerHeartbeatInterval = options.workerHeartbeatInterval;
+      externalStorage = options.externalStorage;
     }
 
     public Builder setNamespace(String namespace) {
@@ -170,6 +173,17 @@ public Builder setWorkerHeartbeatInterval(Duration workerHeartbeatInterval) {
       return this;
     }
 
+    /**
+     * Configures offloading of large payloads to external storage for workflows and activities
+     * created through this client and its workers. When null (the default), external storage is
+     * disabled.
+     */
+    @Experimental
+    public Builder setExternalStorage(ExternalStorageOptions externalStorage) {
+      this.externalStorage = externalStorage;
+      return this;
+    }
+
     public WorkflowClientOptions build() {
       return new WorkflowClientOptions(
           namespace,
@@ -180,7 +194,8 @@ public WorkflowClientOptions build() {
           contextPropagators,
           queryRejectCondition,
           plugins == null ? EMPTY_PLUGINS : plugins,
-          resolveHeartbeatInterval(workerHeartbeatInterval));
+          resolveHeartbeatInterval(workerHeartbeatInterval),
+          externalStorage);
     }
 
     /**
@@ -207,7 +222,8 @@ public WorkflowClientOptions validateAndBuildWithDefaults() {
               ? QueryRejectCondition.QUERY_REJECT_CONDITION_UNSPECIFIED
               : queryRejectCondition,
           plugins == null ? EMPTY_PLUGINS : plugins,
-          resolveHeartbeatInterval(workerHeartbeatInterval));
+          resolveHeartbeatInterval(workerHeartbeatInterval),
+          externalStorage);
     }
 
     private static Duration resolveHeartbeatInterval(Duration raw) {
@@ -250,6 +266,8 @@ private static Duration resolveHeartbeatInterval(Duration raw) {
 
   private final Duration workerHeartbeatInterval;
 
+  private final ExternalStorageOptions externalStorage;
+
   private WorkflowClientOptions(
       String namespace,
       DataConverter dataConverter,
@@ -259,7 +277,8 @@ private WorkflowClientOptions(
       List<ContextPropagator> contextPropagators,
       QueryRejectCondition queryRejectCondition,
       WorkflowClientPlugin[] plugins,
-      Duration workerHeartbeatInterval) {
+      Duration workerHeartbeatInterval,
+      ExternalStorageOptions externalStorage) {
     this.namespace = namespace;
     this.dataConverter = dataConverter;
     this.interceptors = interceptors;
@@ -269,6 +288,7 @@ private WorkflowClientOptions(
     this.queryRejectCondition = queryRejectCondition;
     this.plugins = plugins;
     this.workerHeartbeatInterval = workerHeartbeatInterval;
+    this.externalStorage = externalStorage;
   }
 
   /**
@@ -335,6 +355,12 @@ public Duration getWorkerHeartbeatInterval() {
     return workerHeartbeatInterval;
   }
 
+  /** External storage configuration, or null when external storage is disabled. */
+  @Experimental
+  public ExternalStorageOptions getExternalStorage() {
+    return externalStorage;
+  }
+
   @Override
   public String toString() {
     return "WorkflowClientOptions{"
@@ -359,6 +385,8 @@ public String toString() {
         + Arrays.toString(plugins)
         + ", workerHeartbeatInterval="
         + workerHeartbeatInterval
+        + ", externalStorage="
+        + externalStorage
         + '}';
   }
 
@@ -376,7 +404,8 @@ public boolean equals(Object o) {
         && queryRejectCondition == that.queryRejectCondition
         && Arrays.equals(plugins, that.plugins)
         && com.google.common.base.Objects.equal(
-            workerHeartbeatInterval, that.workerHeartbeatInterval);
+            workerHeartbeatInterval, that.workerHeartbeatInterval)
+        && com.google.common.base.Objects.equal(externalStorage, that.externalStorage);
   }
 
   @Override
@@ -390,6 +419,7 @@ public int hashCode() {
         contextPropagators,
         queryRejectCondition,
         Arrays.hashCode(plugins),
-        workerHeartbeatInterval);
+        workerHeartbeatInterval,
+        externalStorage);
   }
 }

temporal-sdk/src/main/java/io/temporal/common/converter/PayloadAndFailureDataConverter.java

@@ -8,6 +8,7 @@
 import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.failure.v1.Failure;
 import io.temporal.failure.DefaultFailureConverter;
+import io.temporal.internal.payload.storage.ExternalStorageNotConfiguredException;
 import io.temporal.payload.context.SerializationContext;
 import java.lang.reflect.Type;
 import java.util.*;
@@ -71,6 +72,10 @@ public <T> T fromPayload(Payload payload, Class<T> valueClass, Type valueType)
       return (T) new RawValue(payload);
     }
 
+    if (payload.getExternalPayloadsCount() > 0) {
+      throw new ExternalStorageNotConfiguredException();
+    }
+
     try {
       String encoding =
           payload.getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY).toString(UTF_8);

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.ExternalStorageMessageConverter;
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.Arrays;
@@ -21,6 +22,7 @@ public class ActivityExecutionContextFactoryImpl implements ActivityExecutionCon
   private final DataConverter dataConverter;
   private final ScheduledExecutorService heartbeatExecutor;
   private final ManualActivityCompletionClientFactory manualCompletionClientFactory;
+  private final ExternalStorageMessageConverter externalStorage;
   private final ConcurrentMap<ByteBuffer, ActivityExecutionContextImpl> activeContexts =
       new ConcurrentHashMap<>();
 
@@ -31,7 +33,8 @@ public ActivityExecutionContextFactoryImpl(
       Duration maxHeartbeatThrottleInterval,
       Duration defaultHeartbeatThrottleInterval,
       DataConverter dataConverter,
-      ScheduledExecutorService heartbeatExecutor) {
+      ScheduledExecutorService heartbeatExecutor,
+      ExternalStorageMessageConverter externalStorage) {
     this.client = Objects.requireNonNull(client);
     this.identity = identity;
     this.namespace = Objects.requireNonNull(namespace);
@@ -40,9 +43,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 +67,8 @@ public InternalActivityExecutionContext createContext(
             identity,
             maxHeartbeatThrottleInterval,
             defaultHeartbeatThrottleInterval,
-            () -> cleanupContext(info.getTaskToken(), false));
+            () -> cleanupContext(info.getTaskToken(), false),
+            externalStorage);
     activeContexts.put(taskToken, context);
     return context;
   }

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityExecutionContextImpl.java

@@ -10,14 +10,17 @@
 import io.temporal.common.CancellationToken;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
+import io.temporal.internal.payload.storage.ExternalStorageMessageConverter;
 import io.temporal.payload.context.ActivitySerializationContext;
+import io.temporal.payload.storage.StorageDriverActivityInfo;
 import io.temporal.workflow.Functions;
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 
 /**
@@ -55,7 +58,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
       String identity,
       Duration maxHeartbeatThrottleInterval,
       Duration defaultHeartbeatThrottleInterval,
-      Functions.Proc closeCallback) {
+      Functions.Proc closeCallback,
+      @Nullable ExternalStorageMessageConverter externalStorage) {
     this.client = client;
     this.activity = activity;
     this.metricsScope = metricsScope;
@@ -73,7 +77,8 @@ class ActivityExecutionContextImpl implements InternalActivityExecutionContext {
             metricsScope,
             identity,
             maxHeartbeatThrottleInterval,
-            defaultHeartbeatThrottleInterval);
+            defaultHeartbeatThrottleInterval,
+            externalStorage);
   }
 
   /**
@@ -155,7 +160,14 @@ public ManualActivityCompletionClient useLocalManualCompletion() {
           new ActivitySerializationContext(info);
       return new CompletionAwareManualCompletionClient(
           manualCompletionClientFactory.getClient(
-              info.getTaskToken(), metricsScope, activitySerializationContext),
+              info.getTaskToken(),
+              metricsScope,
+              activitySerializationContext,
+              new StorageDriverActivityInfo(
+                  info.getNamespace(),
+                  info.getActivityId(),
+                  info.getActivityRunId(),
+                  info.getActivityType())),
           completionHandle);
     } finally {
       lock.unlock();

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityTaskExecutors.java

@@ -17,6 +17,7 @@
 import io.temporal.common.interceptors.Header;
 import io.temporal.common.interceptors.WorkerInterceptor;
 import io.temporal.internal.common.FailureUtils;
+import io.temporal.internal.payload.storage.ExternalStorageNotConfiguredException;
 import io.temporal.internal.worker.ActivityTaskHandler;
 import io.temporal.payload.context.ActivitySerializationContext;
 import io.temporal.serviceclient.CheckedExceptionWrapper;
@@ -95,6 +96,11 @@ public ActivityTaskHandler.Result execute(ActivityInfoInternal info, Scope metri
         return this.constructSuccessfulResultValue(info, result, dataConverterWithActivityContext);
       } catch (Throwable e) {
         Throwable ex = CheckedExceptionWrapper.unwrap(e);
+        ExternalStorageNotConfiguredException externalStorageFailure =
+            ExternalStorageNotConfiguredException.find(ex);
+        if (externalStorageFailure != null) {
+          throw externalStorageFailure;
+        }
         boolean local = info.isLocal();
         if (ex instanceof ActivityCanceledException) {
           log.info(

temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java

@@ -14,7 +14,10 @@
 import io.temporal.failure.TimeoutFailure;
 import io.temporal.internal.client.ActivityClientHelper;
 import io.temporal.internal.concurrent.structured.CancelSource;
+import io.temporal.internal.payload.storage.ExternalStorageMessageConverter;
 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.lang.reflect.Type;
 import java.time.Duration;
@@ -24,6 +27,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
 import javax.annotation.concurrent.ThreadSafe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -58,6 +62,7 @@ static long getLocalHeartbeatTimeoutBufferMillis() {
   private final long heartbeatIntervalMillis;
   private final DataConverter dataConverter;
   private final DataConverter dataConverterWithActivityContext;
+  private final @Nullable ExternalStorageMessageConverter externalStorage;
 
   private final Scope metricsScope;
   private final Optional<Payloads> prevAttemptHeartbeatDetails;
@@ -89,7 +94,8 @@ public HeartbeatContextImpl(
       Scope metricsScope,
       String identity,
       Duration maxHeartbeatThrottleInterval,
-      Duration defaultHeartbeatThrottleInterval) {
+      Duration defaultHeartbeatThrottleInterval,
+      @Nullable ExternalStorageMessageConverter externalStorage) {
     this(
         service,
         namespace,
@@ -100,6 +106,7 @@ public HeartbeatContextImpl(
         identity,
         maxHeartbeatThrottleInterval,
         defaultHeartbeatThrottleInterval,
+        externalStorage,
         getLocalHeartbeatTimeoutBufferMillis());
   }
 
@@ -113,10 +120,12 @@ public HeartbeatContextImpl(
       String identity,
       Duration maxHeartbeatThrottleInterval,
       Duration defaultHeartbeatThrottleInterval,
+      @Nullable ExternalStorageMessageConverter externalStorage,
       long localHeartbeatTimeoutBufferMillis) {
     this.service = service;
     this.metricsScope = metricsScope;
     this.dataConverter = dataConverter;
+    this.externalStorage = externalStorage;
     this.dataConverterWithActivityContext =
         dataConverter.withContext(
             new ActivitySerializationContext(
@@ -330,6 +339,11 @@ private void checkHeartbeatTimeoutDeadlineLocked() {
     }
   }
 
+  private StorageDriverTargetInfo activityStorageTarget() {
+    return new StorageDriverActivityInfo(
+        namespace, info.getActivityId(), info.getActivityRunId(), info.getActivityType());
+  }
+
   private void sendHeartbeatRequest(Object details) {
     try {
       RecordActivityTaskHeartbeatResponse status =
@@ -339,7 +353,9 @@ private void sendHeartbeatRequest(Object details) {
               identity,
               info.getTaskToken(),
               dataConverterWithActivityContext.toPayloads(details),
-              metricsScope);
+              metricsScope,
+              externalStorage,
+              activityStorageTarget());
       if (status.getCancelRequested()) {
         requestCancelLocked();
       } else if (status.getActivityReset()) {