First pass at NexusSerializationContext for Java

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

@@ -24,12 +24,24 @@ public final class NexusOperationExecutionDescription extends NexusOperationExec
 
   private final DescribeNexusOperationExecutionResponse response;
   private final NexusOperationExecutionInfo info;
-  private final DataConverter dataConverter;
+  private final DataConverter dataConverterWithNexusContext;
+  // User metadata is attached by the caller without a Nexus serialization context, so it has to be
+  // decoded without one too. Everything else on a description belongs to the operation and is
+  // decoded with the operation's context.
+  private final DataConverter contextlessDataConverter;
 
   public NexusOperationExecutionDescription(
       DescribeNexusOperationExecutionResponse response,
       DataConverter dataConverter,
       String namespace) {
+    this(response, dataConverter, dataConverter, namespace);
+  }
+
+  public NexusOperationExecutionDescription(
+      DescribeNexusOperationExecutionResponse response,
+      DataConverter dataConverterWithNexusContext,
+      DataConverter contextlessDataConverter,
+      String namespace) {
     super(
         null,
         response.getInfo().getOperationId(),
@@ -51,7 +63,8 @@ public NexusOperationExecutionDescription(
             : null);
     this.response = response;
     this.info = response.getInfo();
-    this.dataConverter = dataConverter;
+    this.dataConverterWithNexusContext = dataConverterWithNexusContext;
+    this.contextlessDataConverter = contextlessDataConverter;
   }
 
   /** Underlying proto response. Exposed while the Nexus SDK surface is still experimental. */
@@ -124,7 +137,7 @@ public Instant getLastAttemptCompleteTime() {
   @Nullable
   public Exception getLastAttemptFailure() {
     return info.hasLastAttemptFailure()
-        ? dataConverter.failureToException(info.getLastAttemptFailure())
+        ? dataConverterWithNexusContext.failureToException(info.getLastAttemptFailure())
         : null;
   }
 
@@ -140,7 +153,8 @@ public Instant getNextAttemptScheduleTime() {
   @Nullable
   public NexusOperationCancellationInfo getCancellationInfo() {
     return info.hasCancellationInfo()
-        ? new NexusOperationCancellationInfo(info.getCancellationInfo(), dataConverter)
+        ? new NexusOperationCancellationInfo(
+            info.getCancellationInfo(), dataConverterWithNexusContext)
         : null;
   }
 
@@ -183,7 +197,7 @@ public String getStaticSummary() {
     if (!info.hasUserMetadata() || !info.getUserMetadata().hasSummary()) {
       return null;
     }
-    return dataConverter.fromPayload(
+    return contextlessDataConverter.fromPayload(
         info.getUserMetadata().getSummary(), String.class, String.class);
   }
 
@@ -196,7 +210,7 @@ public String getStaticDetails() {
     if (!info.hasUserMetadata() || !info.getUserMetadata().hasDetails()) {
       return null;
     }
-    return dataConverter.fromPayload(
+    return contextlessDataConverter.fromPayload(
         info.getUserMetadata().getDetails(), String.class, String.class);
   }
 
@@ -231,7 +245,7 @@ public <V> Optional<V> getInput(Class<V> valueType, Type genericType) {
       return Optional.empty();
     }
     return Optional.ofNullable(
-        dataConverter.fromPayload(response.getInput(), valueType, genericType));
+        dataConverterWithNexusContext.fromPayload(response.getInput(), valueType, genericType));
   }
 
   /**
@@ -266,7 +280,7 @@ public <V> Optional<V> getResult(Class<V> valueType, Type genericType) {
       return Optional.empty();
     }
     return Optional.ofNullable(
-        dataConverter.fromPayload(response.getResult(), valueType, genericType));
+        dataConverterWithNexusContext.fromPayload(response.getResult(), valueType, genericType));
   }
 
   /**
@@ -275,6 +289,8 @@ public <V> Optional<V> getResult(Class<V> valueType, Type genericType) {
    */
   @Nullable
   public Exception getFailure() {
-    return response.hasFailure() ? dataConverter.failureToException(response.getFailure()) : null;
+    return response.hasFailure()
+        ? dataConverterWithNexusContext.failureToException(response.getFailure())
+        : null;
   }
 }

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

@@ -8,6 +8,7 @@
 import io.temporal.common.interceptors.NexusClientCallsInterceptor.StartNexusOperationExecutionOutput;
 import io.temporal.internal.client.NexusClientResolvedOptions;
 import io.temporal.internal.client.NexusOperationHandleImpl;
+import io.temporal.payload.context.NexusSerializationContext;
 import java.lang.reflect.Type;
 import java.util.Collections;
 import javax.annotation.Nullable;
@@ -43,12 +44,15 @@ class UntypedNexusServiceClientImpl implements UntypedNexusServiceClient {
   @Override
   public UntypedNexusOperationHandle start(
       String operation, StartNexusOperationOptions options, @Nullable Object arg) {
-    Payload payload = serializeInput(arg);
+    Payload payload = serializeInput(arg, operation);
     StartNexusOperationExecutionInput input =
         new StartNexusOperationExecutionInput(
             endpoint, serviceName, operation, payload, options, Collections.emptyMap());
     StartNexusOperationExecutionOutput output = invoker.startNexusOperationExecution(input);
-    return new NexusOperationHandleImpl(output.getOperationId(), output.getRunId(), invoker);
+    // The handle keeps what the start request was for, including when the server returned an
+    // operation that was already running, so the result is decoded the way it was encoded.
+    return new NexusOperationHandleImpl(
+        output.getOperationId(), output.getRunId(), invoker, endpoint, serviceName, operation);
   }
 
   @Override
@@ -71,12 +75,13 @@ public <R> R execute(
     return NexusOperationHandle.fromUntyped(handle, resultClass, resultType).getResult();
   }
 
-  private @Nullable Payload serializeInput(@Nullable Object arg) {
+  private @Nullable Payload serializeInput(@Nullable Object arg, String operation) {
     if (arg == null) {
       return null;
     }
     Class<?> argClass = arg.getClass();
     return dataConverter
+        .withContext(new NexusSerializationContext(endpoint, serviceName, operation))
         .toPayload(arg)
         .orElseThrow(
             () ->

temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusClientCallsInterceptor.java

@@ -10,6 +10,7 @@
 import io.temporal.client.NexusOperationHandle;
 import io.temporal.client.StartNexusOperationOptions;
 import io.temporal.common.Experimental;
+import io.temporal.payload.context.NexusSerializationContext;
 import java.lang.reflect.Type;
 import java.util.Collections;
 import java.util.Map;
@@ -250,18 +251,65 @@ final class GetNexusOperationResultInput<R> {
     private final @Nonnull Deadline deadline;
     private final Class<R> resultClass;
     private final @Nullable Type resultType;
+    private final @Nullable String endpoint;
+    private final @Nullable String service;
+    private final @Nullable String operation;
 
+    /**
+     * Equivalent to {@link #GetNexusOperationResultInput(String, String, Deadline, Class, Type,
+     * String, String, String)} with no endpoint, service or operation, which is the case for a
+     * handle obtained by operation ID rather than by starting an operation.
+     */
     public GetNexusOperationResultInput(
         String operationId,
         @Nullable String runId,
         @Nonnull Deadline deadline,
         Class<R> resultClass,
         @Nullable Type resultType) {
+      this(operationId, runId, deadline, resultClass, resultType, null, null, null);
+    }
+
+    /**
+     * The endpoint, service and operation identify the Nexus operation the result is being read
+     * for, and are used to build the {@link NexusSerializationContext} the result and failure are
+     * decoded with. They must all be set or all be {@code null}: a partially identified operation
+     * would silently decode without a context, which for a converter that varies by context means
+     * reading the payload the wrong way rather than failing.
+     *
+     * @param endpoint Nexus endpoint the operation was started on, or {@code null} if the operation
+     *     was not started through this handle
+     * @param service Nexus service the operation was started on, or {@code null}
+     * @param operation Nexus operation that was started, or {@code null}
+     * @throws IllegalArgumentException if only some of endpoint, service and operation are set
+     */
+    public GetNexusOperationResultInput(
+        String operationId,
+        @Nullable String runId,
+        @Nonnull Deadline deadline,
+        Class<R> resultClass,
+        @Nullable Type resultType,
+        @Nullable String endpoint,
+        @Nullable String service,
+        @Nullable String operation) {
+      boolean anySet = endpoint != null || service != null || operation != null;
+      boolean allSet = endpoint != null && service != null && operation != null;
+      if (anySet && !allSet) {
+        throw new IllegalArgumentException(
+            "endpoint, service and operation must all be set or all be null, got endpoint="
+                + endpoint
+                + ", service="
+                + service
+                + ", operation="
+                + operation);
+      }
       this.operationId = operationId;
       this.runId = runId;
       this.deadline = deadline;
       this.resultClass = resultClass;
       this.resultType = resultType;
+      this.endpoint = endpoint;
+      this.service = service;
+      this.operation = operation;
     }
 
     public String getOperationId() {
@@ -285,6 +333,34 @@ public Class<R> getResultClass() {
     public Type getResultType() {
       return resultType;
     }
+
+    /**
+     * Nexus endpoint the operation was started on. {@code null} when the operation was not started
+     * through this handle, in which case {@link #getService()} and {@link #getOperation()} are
+     * {@code null} too and the result is decoded without a Nexus serialization context.
+     */
+    @Nullable
+    public String getEndpoint() {
+      return endpoint;
+    }
+
+    /**
+     * Nexus service the operation was started on, or {@code null}. Set exactly when {@link
+     * #getEndpoint()} is set.
+     */
+    @Nullable
+    public String getService() {
+      return service;
+    }
+
+    /**
+     * Nexus operation that was started, or {@code null}. Set exactly when {@link #getEndpoint()} is
+     * set.
+     */
+    @Nullable
+    public String getOperation() {
+      return operation;
+    }
   }
 
   final class GetNexusOperationResultOutput<R> {

temporal-sdk/src/main/java/io/temporal/internal/client/NexusOperationHandleImpl.java

@@ -25,9 +25,25 @@ public final class NexusOperationHandleImpl implements UntypedNexusOperationHand
   private final String operationId;
   private final @Nullable String runId;
   private final NexusClientCallsInterceptor interceptor;
+  // What the operation was started on, retained so its result and failure are decoded with the
+  // same serialization context that encoded them. All null for a handle obtained by operation ID,
+  // which never saw a start request.
+  private final @Nullable String endpoint;
+  private final @Nullable String service;
+  private final @Nullable String operation;
 
   public NexusOperationHandleImpl(
       String operationId, @Nullable String runId, NexusClientCallsInterceptor interceptor) {
+    this(operationId, runId, interceptor, null, null, null);
+  }
+
+  public NexusOperationHandleImpl(
+      String operationId,
+      @Nullable String runId,
+      NexusClientCallsInterceptor interceptor,
+      @Nullable String endpoint,
+      @Nullable String service,
+      @Nullable String operation) {
     if (operationId == null) {
       throw new IllegalArgumentException("operationId is required");
     }
@@ -37,6 +53,9 @@ public NexusOperationHandleImpl(
     this.operationId = operationId;
     this.runId = runId;
     this.interceptor = interceptor;
+    this.endpoint = endpoint;
+    this.service = service;
+    this.operation = operation;
   }
 
   @Override
@@ -116,7 +135,14 @@ public <R> R getResult(
       throws TimeoutException {
     GetNexusOperationResultInput<R> input =
         new GetNexusOperationResultInput<>(
-            operationId, runId, Deadline.after(timeout, unit), resultClass, resultType);
+            operationId,
+            runId,
+            Deadline.after(timeout, unit),
+            resultClass,
+            resultType,
+            endpoint,
+            service,
+            operation);
     return interceptor.getNexusOperationResult(input).getResult();
   }
 
@@ -131,7 +157,14 @@ public <R> CompletableFuture<R> getResultAsync(
       long timeout, TimeUnit unit, Class<R> resultClass, @Nullable Type resultType) {
     GetNexusOperationResultInput<R> input =
         new GetNexusOperationResultInput<>(
-            operationId, runId, Deadline.after(timeout, unit), resultClass, resultType);
+            operationId,
+            runId,
+            Deadline.after(timeout, unit),
+            resultClass,
+            resultType,
+            endpoint,
+            service,
+            operation);
     return interceptor
         .getNexusOperationResultAsync(input)
         .thenApply(GetNexusOperationResultOutput::getResult);