Add RawValue support (#2492)

Add RawValue support

Author: Quinn-With-Two-Ns

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

@@ -66,6 +66,10 @@
  *
  * A {@link DataConverter} created on previous step may be bundled with {@link PayloadCodec}s using
  * {@link CodecDataConverter} or used directly if no custom {@link PayloadCodec}s are needed.
+ *
+ * <p>{@link DataConverter} is expected to pass the {@link RawValue} payload through without
+ * conversion. Though it should still apply the {@link PayloadCodec} to the {@link RawValue}
+ * payloads.
  */
 public interface DataConverter {
 

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

@@ -27,8 +27,6 @@
 /**
  * A {@link DataConverter} that delegates payload conversion to type specific {@link
  * PayloadConverter} instances, and delegates failure conversions to a {@link FailureConverter}.
- *
- * @author fateev
  */
 public class DefaultDataConverter extends PayloadAndFailureDataConverter {
 

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

@@ -65,6 +65,12 @@ public PayloadAndFailureDataConverter(@Nonnull List<PayloadConverter> converters
 
   @Override
   public <T> Optional<Payload> toPayload(T value) throws DataConverterException {
+    // Raw values payload should be passed through without conversion
+    if (value instanceof RawValue) {
+      RawValue rv = (RawValue) value;
+      return Optional.of(rv.getPayload());
+    }
+
     for (PayloadConverter converter : converters) {
       Optional<Payload> result =
           (serializationContext != null ? converter.withContext(serializationContext) : converter)
@@ -77,9 +83,14 @@ public <T> Optional<Payload> toPayload(T value) throws DataConverterException {
         "No PayloadConverter is registered with this DataConverter that accepts value:" + value);
   }
 
+  @SuppressWarnings("unchecked")
   @Override
   public <T> T fromPayload(Payload payload, Class<T> valueClass, Type valueType)
       throws DataConverterException {
+    if (valueClass == RawValue.class) {
+      return (T) new RawValue(payload);
+    }
+
     try {
       String encoding =
           payload.getMetadataOrThrow(EncodingKeys.METADATA_ENCODING_KEY).toString(UTF_8);

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

@@ -33,7 +33,9 @@
  * Used by the framework to serialize/deserialize method parameters that need to be sent over the
  * wire.
  *
- * @author fateev
+ * <p>{@link PayloadConverter} is expected to pass the {@link RawValue} payload through without
+ * conversion. Though it should still apply the {@link io.temporal.payload.codec.PayloadCodec} to
+ * the {@link RawValue} payloads.
  */
 public interface PayloadConverter {
 

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.common.converter;
+
+import io.temporal.api.common.v1.Payload;
+import java.util.Objects;
+
+/**
+ * RawValue is a representation of an unconverted, raw payload.
+ *
+ * <p>This type can be used as a parameter or return type in workflows and activities to pass
+ * through a raw payload. Encoding/decoding of the payload is still done by the system.
+ */
+public final class RawValue {
+  private final Payload payload;
+
+  public RawValue(Payload payload) {
+    this.payload = Objects.requireNonNull(payload);
+  }
+
+  public Payload getPayload() {
+    return payload;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    RawValue rawValue = (RawValue) o;
+    return Objects.equals(payload, rawValue.payload);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(payload);
+  }
+
+  @Override
+  public String toString() {
+    return "RawValue{" + "payload=" + payload + '}';
+  }
+}

temporal-sdk/src/test/java/io/temporal/common/converter/CodecDataConverterTest.java

@@ -26,13 +26,15 @@
 
 import com.google.protobuf.ByteString;
 import io.temporal.api.common.v1.Payload;
+import io.temporal.api.common.v1.Payloads;
 import io.temporal.api.failure.v1.Failure;
 import io.temporal.failure.ApplicationFailure;
 import io.temporal.failure.TemporalFailure;
 import io.temporal.payload.codec.PayloadCodec;
 import io.temporal.payload.codec.PayloadCodecException;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import org.junit.Before;
@@ -127,6 +129,16 @@ public void testDetailsAreEncoded() {
     assertArrayEquals(new int[] {1, 2, 3}, decodedDetailsPayloads.get(2, int[].class, int[].class));
   }
 
+  @Test
+  public void testRawValuePassThrough() {
+    Payload p = Payload.newBuilder().setData(ByteString.copyFromUtf8("test")).build();
+    Optional<Payloads> data = dataConverter.toPayloads(new RawValue(p));
+    // Assert that the payload is still encoded
+    assertTrue(isEncoded(data.get().getPayloads(0)));
+    RawValue converted = dataConverter.fromPayloads(0, data, RawValue.class, RawValue.class);
+    assertEquals(p, converted.getPayload());
+  }
+
   static boolean isEncoded(Payload payload) {
     return payload.getData().startsWith(PrefixPayloadCodec.PREFIX);
   }

temporal-sdk/src/test/java/io/temporal/common/converter/ProtoPayloadConverterTest.java

@@ -66,6 +66,31 @@ public void testProto() {
     assertEquals(execution, converted);
   }
 
+  @Test
+  public void testRawValue() {
+    DataConverter converter = DefaultDataConverter.STANDARD_INSTANCE;
+    ProtobufPayloadConverter protoConverter = new ProtobufPayloadConverter();
+    WorkflowExecution execution =
+        WorkflowExecution.newBuilder()
+            .setWorkflowId(UUID.randomUUID().toString())
+            .setRunId(UUID.randomUUID().toString())
+            .build();
+    Optional<Payloads> data =
+        converter.toPayloads(new RawValue(protoConverter.toData(execution).get()));
+    WorkflowExecution converted =
+        converter.fromPayloads(0, data, WorkflowExecution.class, WorkflowExecution.class);
+    assertEquals(execution, converted);
+  }
+
+  @Test
+  public void testRawValuePassThrough() {
+    DataConverter converter = DefaultDataConverter.STANDARD_INSTANCE;
+    Payload p = Payload.newBuilder().setData(ByteString.copyFromUtf8("test")).build();
+    Optional<Payloads> data = converter.toPayloads(new RawValue(p));
+    RawValue converted = converter.fromPayloads(0, data, RawValue.class, RawValue.class);
+    assertEquals(p, converted.getPayload());
+  }
+
   @Test
   public void testCustomProto() {
     DataConverter converter =

temporal-sdk/src/test/java/io/temporal/workflow/activityTests/TestRawValueActivity.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
+ *
+ * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this material except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.temporal.workflow.activityTests;
+
+import static org.junit.Assert.*;
+
+import com.google.protobuf.ByteString;
+import io.temporal.activity.ActivityOptions;
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.converter.RawValue;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.Workflow;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+import io.temporal.workflow.shared.TestActivities.TestActivitiesImpl;
+import io.temporal.workflow.shared.TestActivities.VariousTestActivities;
+import java.time.Duration;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class TestRawValueActivity {
+
+  private final TestActivitiesImpl activitiesImpl = new TestActivitiesImpl();
+
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(RawValueTestWorkflowImpl.class)
+          .setActivityImplementations(activitiesImpl)
+          .build();
+
+  @Test
+  public void testRawValueEndToEnd() {
+    RawValueTestWorkflow workflowStub =
+        testWorkflowRule.newWorkflowStubTimeoutOptions(RawValueTestWorkflow.class);
+    // Intentionally don't set an encoding to test that the payload is passed through as is.
+    Payload p = Payload.newBuilder().setData(ByteString.copyFromUtf8("test")).build();
+    RawValue input = new RawValue(p);
+    RawValue result = workflowStub.execute(input);
+    Assert.assertEquals(input.getPayload(), result.getPayload());
+  }
+
+  @WorkflowInterface
+  public interface RawValueTestWorkflow {
+    @WorkflowMethod
+    RawValue execute(RawValue value);
+  }
+
+  public static class RawValueTestWorkflowImpl implements RawValueTestWorkflow {
+    @Override
+    public RawValue execute(RawValue value) {
+      ActivityOptions options =
+          ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(5)).build();
+      VariousTestActivities activities =
+          Workflow.newActivityStub(VariousTestActivities.class, options);
+      return activities.rawValueActivity(value);
+    }
+  }
+}

temporal-sdk/src/test/java/io/temporal/workflow/shared/TestActivities.java

@@ -35,6 +35,7 @@
 import io.temporal.client.ActivityCompletionClient;
 import io.temporal.client.ActivityNotExistsException;
 import io.temporal.common.MethodRetry;
+import io.temporal.common.converter.RawValue;
 import io.temporal.failure.ApplicationFailure;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import java.io.Closeable;
@@ -145,6 +146,8 @@ public interface VariousTestActivities {
     void throwIOAnnotated();
 
     List<UUID> activityUUIDList(List<UUID> arg);
+
+    RawValue rawValueActivity(RawValue arg);
   }
 
   @ActivityInterface
@@ -411,6 +414,11 @@ public List<UUID> activityUUIDList(List<UUID> arg) {
       return arg;
     }
 
+    @Override
+    public RawValue rawValueActivity(RawValue arg) {
+      return arg;
+    }
+
     public int getLastAttempt() {
       return lastAttempt;
     }