External Storage Integration: Activity worker, client

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);

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. */

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> T getMemo(String key, Class<T> 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

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,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();