External Storage Integration: Activity worker, client

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.ExternalStorageRunner;
 import io.temporal.internal.sync.StubMarker;
 import io.temporal.internal.worker.HeartbeatManager;
+import io.temporal.payload.storage.ExternalStorage;
 import io.temporal.serviceclient.MetricsTag;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.serviceclient.WorkflowServiceStubsPlugin;
@@ -56,6 +59,7 @@ final class WorkflowClientInternalImpl implements WorkflowClient, WorkflowClient
   private final WorkerFactoryRegistry workerFactoryRegistry = new WorkerFactoryRegistry();
   private final String workerGroupingKey = java.util.UUID.randomUUID().toString();
   private final @Nullable HeartbeatManager heartbeatManager;
+  private final @Nullable ExternalStorageRunner externalStorage;
 
   /**
    * Creates client that connects to an instance of the Temporal Service. Cannot be used from within
@@ -106,15 +110,27 @@ public static WorkflowClient newInstance(
             .getOptions()
             .getMetricsScope()
             .tagged(MetricsTag.defaultTags(options.getNamespace()));
-    this.genericClient = new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
+    ExternalStorage externalStorageConfig = options.getDataConverter().getExternalStorage();
+    ExternalStorageRunner externalStorage =
+        externalStorageConfig == null ? null : ExternalStorageRunner.create(externalStorageConfig);
+    this.externalStorage = externalStorage;
+    GenericWorkflowClient genericClient =
+        new GenericWorkflowClientImpl(workflowServiceStubs, metricsScope);
+    if (externalStorage != null) {
+      genericClient =
+          new ExternalStorageGenericWorkflowClient(
+              genericClient, externalStorage, options.getNamespace());
+    }
+    this.genericClient = genericClient;
     this.interceptors = options.getInterceptors();
     this.workflowClientCallsInvoker = initializeClientInvoker();
     this.manualActivityCompletionClientFactory =
         ManualActivityCompletionClientFactory.newFactory(
             workflowServiceStubs,
             options.getNamespace(),
             options.getIdentity(),
-            options.getDataConverter());
+            options.getDataConverter(),
+            externalStorage);
 
     java.time.Duration heartbeatInterval = options.getWorkerHeartbeatInterval();
     if (!heartbeatInterval.isNegative()) {
@@ -815,6 +831,12 @@ public HeartbeatManager getHeartbeatManager() {
     return heartbeatManager;
   }
 
+  @Override
+  @Nullable
+  public ExternalStorageRunner getExternalStorage() {
+    return externalStorage;
+  }
+
   @Override
   public NexusStartWorkflowResponse startNexus(
       NexusStartWorkflowRequest request, Functions.Proc workflow) {

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

@@ -1,5 +1,6 @@
 package io.temporal.client;
 
+import io.temporal.api.common.v1.Payload;
 import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionResponse;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.payload.context.WorkflowSerializationContext;
@@ -29,15 +30,15 @@ public String getStaticSummary() {
     if (!response.getExecutionConfig().getUserMetadata().hasSummary()) {
       return null;
     }
+    Payload summary =
+        resolveExternalStorageReference(
+            response.getExecutionConfig().getUserMetadata().getSummary());
     return dataConverter
         .withContext(
             new WorkflowSerializationContext(
                 response.getWorkflowExecutionInfo().getParentNamespaceId(),
                 response.getWorkflowExecutionInfo().getExecution().getWorkflowId()))
-        .fromPayload(
-            response.getExecutionConfig().getUserMetadata().getSummary(),
-            String.class,
-            String.class);
+        .fromPayload(summary, String.class, String.class);
   }
 
   /**
@@ -51,15 +52,15 @@ public String getStaticDetails() {
     if (!response.getExecutionConfig().getUserMetadata().hasDetails()) {
       return null;
     }
+    Payload details =
+        resolveExternalStorageReference(
+            response.getExecutionConfig().getUserMetadata().getDetails());
     return dataConverter
         .withContext(
             new WorkflowSerializationContext(
                 response.getWorkflowExecutionInfo().getParentNamespaceId(),
                 response.getWorkflowExecutionInfo().getExecution().getWorkflowId()))
-        .fromPayload(
-            response.getExecutionConfig().getUserMetadata().getDetails(),
-            String.class,
-            String.class);
+        .fromPayload(details, String.class, String.class);
   }
 
   /** Returns the raw response from the Temporal service. */

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

@@ -2,14 +2,18 @@
 
 import com.google.common.base.Preconditions;
 import io.temporal.api.common.v1.Payload;
+import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.api.enums.v1.WorkflowExecutionStatus;
 import io.temporal.api.workflow.v1.WorkflowExecutionInfo;
 import io.temporal.common.SearchAttributes;
 import io.temporal.common.converter.DataConverter;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.internal.common.SearchAttributesUtil;
+import io.temporal.internal.payload.storage.ExternalStorageReferences;
+import io.temporal.internal.payload.storage.ExternalStorageRunner;
 import io.temporal.payload.context.WorkflowSerializationContext;
+import io.temporal.payload.storage.ExternalStorage;
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.time.Instant;
@@ -123,13 +127,29 @@ public <T> T getMemo(String key, Class<T> valueClass, Type genericType) {
     if (memo == null) {
       return null;
     }
+    memo = resolveExternalStorageReference(memo);
     return dataConverter
         .withContext(
             new WorkflowSerializationContext(
                 info.getParentNamespaceId(), info.getExecution().getWorkflowId()))
         .fromPayload(memo, valueClass, genericType);
   }
 
+  /**
+   * Resolves an external-storage reference payload to its stored contents, lazily, when a getter
+   * reads it. Uses the external storage attached to this result's data converter, or returns the
+   * payload unchanged when it is not a reference or no external storage is configured.
+   */
+  protected Payload resolveExternalStorageReference(Payload payload) {
+    ExternalStorage externalStorage = dataConverter.getExternalStorage();
+    if (externalStorage == null || !ExternalStorageReferences.isReference(payload)) {
+      return payload;
+    }
+    return ExternalStorageRunner.create(externalStorage)
+        .retrieveBlocking(Payloads.newBuilder().addPayloads(payload).build())
+        .getPayloads(0);
+  }
+
   @Nonnull
   public WorkflowExecutionInfo getWorkflowExecutionInfo() {
     return info;

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

@@ -4,13 +4,15 @@
 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;
 import java.util.Objects;
 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<ByteBuffer, ActivityExecutionContextImpl> 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;
   }

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.ExternalStorageRunner;
 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 ExternalStorageRunner 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();