External Storage Integration: Lazy resolving references, general refactoring

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

@@ -12,6 +12,7 @@
 import io.temporal.payload.codec.ChainCodec;
 import io.temporal.payload.codec.PayloadCodec;
 import io.temporal.payload.context.SerializationContext;
+import io.temporal.payload.storage.ExternalStorage;
 import java.lang.reflect.Type;
 import java.util.Collection;
 import java.util.Collections;
@@ -190,6 +191,12 @@ public CodecDataConverter withContext(@Nonnull SerializationContext context) {
     return new CodecDataConverter(dataConverter, chainCodec, encodeFailureAttributes, context);
   }
 
+  @Override
+  @Nullable
+  public ExternalStorage getExternalStorage() {
+    return dataConverter.getExternalStorage();
+  }
+
   @Nonnull
   @Override
   public List<Payload> encode(@Nonnull List<Payload> payloads) {

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

@@ -11,10 +11,12 @@
 import io.temporal.failure.DefaultFailureConverter;
 import io.temporal.payload.codec.PayloadCodec;
 import io.temporal.payload.context.SerializationContext;
+import io.temporal.payload.storage.ExternalStorage;
 import java.lang.reflect.Type;
 import java.util.Arrays;
 import java.util.Optional;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 /**
  * Used by the framework to serialize/deserialize method parameters that need to be sent over the
@@ -202,6 +204,16 @@ default DataConverter withContext(@Nonnull SerializationContext context) {
     return this;
   }
 
+  /**
+   * External storage offloads large payloads. This should not be used from inside workflow code as
+   * it performs nondeterministic operations.
+   */
+  @Experimental
+  @Nullable
+  default ExternalStorage getExternalStorage() {
+    return null;
+  }
+
   /**
    * @deprecated use {@link DataConverter#fromPayloads(int, Optional, Class, Type)}. This is an SDK
    *     implementation detail and never was expected to be exposed to users.

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

@@ -1,8 +1,10 @@
 package io.temporal.common.converter;
 
 import com.google.common.base.Preconditions;
+import io.temporal.payload.storage.ExternalStorage;
 import java.util.*;
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 
 /**
  * A {@link DataConverter} that delegates payload conversion to type specific {@link
@@ -101,4 +103,13 @@ public DefaultDataConverter withFailureConverter(@Nonnull FailureConverter failu
     this.failureConverter = Preconditions.checkNotNull(failureConverter, "failureConverter");
     return this;
   }
+
+  /**
+   * Modifies this {@code DefaultDataConverter} by attaching external storage, used to offload and
+   * restore large payloads at task and RPC boundaries.
+   */
+  public DefaultDataConverter withExternalStorage(@Nullable ExternalStorage externalStorage) {
+    this.externalStorage = externalStorage;
+    return this;
+  }
 }

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

@@ -8,7 +8,10 @@
 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.ExternalStorageReferences;
+import io.temporal.internal.payload.storage.ExternalStorageUnhandledReferenceException;
 import io.temporal.payload.context.SerializationContext;
+import io.temporal.payload.storage.ExternalStorage;
 import java.lang.reflect.Type;
 import java.util.*;
 import javax.annotation.Nonnull;
@@ -22,6 +25,7 @@ class PayloadAndFailureDataConverter implements DataConverter {
   volatile List<PayloadConverter> converters;
   volatile Map<String, PayloadConverter> convertersMap;
   volatile FailureConverter failureConverter;
+  volatile @Nullable ExternalStorage externalStorage;
   private final @Nullable SerializationContext serializationContext;
 
   public PayloadAndFailureDataConverter(@Nonnull List<PayloadConverter> converters) {
@@ -71,6 +75,10 @@ public <T> T fromPayload(Payload payload, Class<T> valueClass, Type valueType)
       return (T) new RawValue(payload);
     }
 
+    if (ExternalStorageReferences.isReference(payload)) {
+      throw new ExternalStorageUnhandledReferenceException();
+    }
+
     try {
       String encoding =
           payload.getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY).toString(UTF_8);
@@ -143,9 +151,18 @@ public Failure exceptionToFailure(@Nonnull Throwable throwable) {
         .exceptionToFailure(throwable, this);
   }
 
+  @Override
+  @Nullable
+  public ExternalStorage getExternalStorage() {
+    return externalStorage;
+  }
+
   @Override
   public @Nonnull DataConverter withContext(@Nonnull SerializationContext context) {
-    return new PayloadAndFailureDataConverter(converters, convertersMap, failureConverter, context);
+    PayloadAndFailureDataConverter copy =
+        new PayloadAndFailureDataConverter(converters, convertersMap, failureConverter, context);
+    copy.externalStorage = this.externalStorage;
+    return copy;
   }
 
   static Map<String, PayloadConverter> createConvertersMap(List<PayloadConverter> converters) {

temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageNotConfiguredException.java

@@ -0,0 +1,18 @@
+package io.temporal.internal.payload.storage;
+
+import io.temporal.common.converter.DataConverterException;
+
+/**
+ * Signals that a payload referenced in external storage needs to be retrieved, but external storage
+ * is not configured. Logs a TMPRL1105 error.
+ */
+public final class ExternalStorageNotConfiguredException extends DataConverterException {
+  public ExternalStorageNotConfiguredException() {
+    super(
+        "[TMPRL1105] Encountered a reference to a payload in external storage, but no external "
+            + "storage is configured to retrieve it. Configure external storage on the "
+            + "DataConverter with "
+            + "DefaultDataConverter.withExternalStorage(...) and provide a driver able to "
+            + "retrieve it.");
+  }
+}

temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStoragePayloadTransformer.java

@@ -4,7 +4,7 @@
 import io.temporal.common.CancellationToken;
 import io.temporal.internal.common.ListUtils;
 import io.temporal.internal.concurrent.structured.TaskScope;
-import io.temporal.payload.storage.ExternalStorageOptions;
+import io.temporal.payload.storage.ExternalStorage;
 import io.temporal.payload.storage.StorageDriver;
 import io.temporal.payload.storage.StorageDriverClaim;
 import io.temporal.payload.storage.StorageDriverRetrieveContext;
@@ -30,7 +30,7 @@ final class ExternalStoragePayloadTransformer {
   private final StorageDriverSelector selector;
   private final int payloadSizeThreshold;
 
-  static ExternalStoragePayloadTransformer fromOptions(ExternalStorageOptions options) {
+  static ExternalStoragePayloadTransformer fromOptions(ExternalStorage options) {
     Map<String, StorageDriver> driversByName = new LinkedHashMap<>();
     for (StorageDriver driver : options.getDrivers()) {
       driversByName.put(driver.getName(), driver);

temporal-sdk/src/main/java/io/temporal/internal/payload/storage/ExternalStorageReferences.java

@@ -9,7 +9,7 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
-final class ExternalStorageReferences {
+public final class ExternalStorageReferences {
   private static final String ENCODING_PROTOBUF_JSON = "json/protobuf";
   private static final String REFERENCE_MESSAGE_TYPE =
       ExternalStorageReference.getDescriptor().getFullName();
@@ -64,8 +64,7 @@ static Payload toReferencePayload(
    * producer that omits it still yields a readable reference.
    */
   static @Nullable ParsedReference tryParseReference(@Nonnull Payload payload) {
-    if (!hasMetadata(payload, EncodingKeys.METADATA_ENCODING_KEY, ENCODING_PROTOBUF_JSON)
-        || !hasMetadata(payload, EncodingKeys.METADATA_MESSAGE_TYPE_KEY, REFERENCE_MESSAGE_TYPE)) {
+    if (!isReference(payload)) {
       return null;
     }
     ExternalStorageReference.Builder builder = ExternalStorageReference.newBuilder();
@@ -79,6 +78,12 @@ static Payload toReferencePayload(
         reference.getDriverName(), new StorageDriverClaim(reference.getClaimDataMap()));
   }
 
+  /** True if {@code payload} has an external storage reference encoding and message type. */
+  public static boolean isReference(Payload payload) {
+    return hasMetadata(payload, EncodingKeys.METADATA_ENCODING_KEY, ENCODING_PROTOBUF_JSON)
+        && hasMetadata(payload, EncodingKeys.METADATA_MESSAGE_TYPE_KEY, REFERENCE_MESSAGE_TYPE);
+  }
+
   private static boolean hasMetadata(Payload payload, String key, String expected) {
     ByteString value = payload.getMetadataMap().get(key);
     return value != null && expected.equals(value.toStringUtf8());