address comments 2: workflow exec overloads

Author: mavemuri

.github/workflows/ci.yml

@@ -120,6 +120,8 @@ jobs:
               --dynamic-config-value history.enableChasm=true \
               --dynamic-config-value history.enableCHASMSignalBacklinks=true \
               --dynamic-config-value history.enableTransitionHistory=true \
+              --dynamic-config-value history.enableUpdateCallbacks=true \
+              --dynamic-config-value history.enableCHASMCallbacks=true \
               --dynamic-config-value frontend.enableCancelWorkerPollsOnShutdown=true \
               --dynamic-config-value frontend.workerCommandsEnabled=true \
               --dynamic-config-value system.enableCancelActivityWorkerCommand=true &

temporal-sdk/src/main/java/io/temporal/nexus/CancelUpdateWorkflowInput.java

@@ -1,6 +1,5 @@
 package io.temporal.nexus;
 
-import com.google.common.base.Strings;
 import io.temporal.common.Experimental;
 import java.util.Objects;
 
@@ -17,7 +16,7 @@ public final class CancelUpdateWorkflowInput {
 
   public CancelUpdateWorkflowInput(String workflowId, String runId, String updateId) {
     this.workflowId = Objects.requireNonNull(workflowId);
-    this.runId = Strings.nullToEmpty(runId);
+    this.runId = Objects.requireNonNull(runId);
     this.updateId = Objects.requireNonNull(updateId);
   }
 

temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClient.java

@@ -1,6 +1,7 @@
 package io.temporal.nexus;
 
 import io.nexusrpc.OperationException;
+import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.client.UpdateOptions;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowOptions;
@@ -510,7 +511,6 @@ <R> TemporalOperationResult<R> startWorkflow(
       WorkflowOptions options,
       Object... args);
 
-  // draft-review: all these combinations are for 1-1 compatibility - check if they can be trimmed
   /**
    * Starts a workflow update on an existing workflow as a Nexus operation. The result is delivered
    * asynchronously via the Nexus completion callback, unless the update RPC comes back already
@@ -746,7 +746,45 @@ <T, A1, A2, A3, A4, A5, A6, R> TemporalOperationResult<R> startWorkflowUpdate(
       UpdateOptions<R> options)
       throws OperationException;
 
-  // draft-review: check if all these are also necessary to preserve 1:1 compatibility
+  /**
+   * TBD: determine if all the other 14 func+proc+void are required and add if needed Starts a
+   * six-argument workflow update on the provided workflow execution. See {@link
+   * #startWorkflowUpdate(Class, String, Functions.Func1, UpdateOptions)} for the full behavior
+   * contract.
+   *
+   * @param <T>
+   * @param <A1>
+   * @param <A2>
+   * @param <A3>
+   * @param <A4>
+   * @param <A5>
+   * @param <A6>
+   * @param <R>
+   * @param workflowClass
+   * @param execution
+   * @param updateMethod
+   * @param arg1
+   * @param arg2
+   * @param arg3
+   * @param arg4
+   * @param arg5
+   * @param arg6
+   * @param options
+   * @return
+   * @throws OperationException
+   */
+  <T, A1, A2, A3, A4, A5, A6, R> TemporalOperationResult<R> startWorkflowUpdate(
+      Class<T> workflowClass,
+      WorkflowExecution execution,
+      Functions.Func7<T, A1, A2, A3, A4, A5, A6, R> updateMethod,
+      A1 arg1,
+      A2 arg2,
+      A3 arg3,
+      A4 arg4,
+      A5 arg5,
+      A6 arg6,
+      UpdateOptions<R> options)
+      throws OperationException;
 
   /**
    * Starts a zero-argument workflow update with no return value on an existing workflow as a Nexus

temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java

@@ -4,12 +4,15 @@
 import com.google.common.base.Strings;
 import io.nexusrpc.OperationException;
 import io.nexusrpc.handler.HandlerException;
+import io.nexusrpc.handler.HandlerException.RetryBehavior;
 import io.nexusrpc.handler.OperationContext;
 import io.nexusrpc.handler.OperationStartDetails;
+import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.client.UpdateOptions;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowOptions;
 import io.temporal.client.WorkflowStub;
+import io.temporal.client.WorkflowTargetOptions;
 import io.temporal.client.WorkflowUpdateException;
 import io.temporal.client.WorkflowUpdateHandle;
 import io.temporal.client.WorkflowUpdateStage;
@@ -596,15 +599,21 @@ private <R> TemporalOperationResult<R> executeUpdate(
   private <R> void checkNexusUpdateOptionsValid(UpdateOptions<R> options)
       throws OperationException {
     if (options.getWaitForStage() != WorkflowUpdateStage.ACCEPTED) {
-      throw OperationException.failed(
-          "workflow update Nexus operation only support WaitForStage Accepted");
+      throw new HandlerException(
+          HandlerException.ErrorType.BAD_REQUEST,
+          "invalid update request",
+          new IllegalArgumentException(
+              "nexus op workflow updates only support WorkflowUpdateStageAccepted for async updates"),
+          RetryBehavior.RETRYABLE);
     }
-    // draft-review: TBD, this is still under discussion as we may want to let
-    // handlers trigger invalid updates in some cases and just retry forever
     try {
       options.validate();
     } catch (IllegalStateException e) {
-      throw OperationException.failed(e);
+      throw new HandlerException(
+          HandlerException.ErrorType.INTERNAL,
+          "invalid update request",
+          e,
+          RetryBehavior.RETRYABLE);
     }
   }
 
@@ -617,4 +626,28 @@ private void markAsyncOperationStarted() {
                   + "Use getWorkflowClient() for additional workflow interactions."));
     }
   }
+
+  @Override
+  public <T, A1, A2, A3, A4, A5, A6, R> TemporalOperationResult<R> startWorkflowUpdate(
+      Class<T> workflowClass,
+      WorkflowExecution execution,
+      Functions.Func7<T, A1, A2, A3, A4, A5, A6, R> updateMethod,
+      A1 arg1,
+      A2 arg2,
+      A3 arg3,
+      A4 arg4,
+      A5 arg5,
+      A6 arg6,
+      UpdateOptions<R> options)
+      throws OperationException {
+    T stub =
+        client.newWorkflowStub(
+            workflowClass,
+            WorkflowTargetOptions.newBuilder().setWorkflowExecution(execution).build());
+    return executeUpdate(
+        options,
+        effective ->
+            WorkflowClient.startUpdate(
+                () -> updateMethod.apply(stub, arg1, arg2, arg3, arg4, arg5, arg6), effective));
+  }
 }

temporal-sdk/src/test/java/io/temporal/workflow/nexus/UpdateWorkflowOperationTest.java

@@ -14,7 +14,6 @@
 import io.temporal.client.WorkflowUpdateStage;
 import io.temporal.failure.ApplicationFailure;
 import io.temporal.failure.NexusOperationFailure;
-import io.temporal.internal.common.env.EnvironmentVariableUtils;
 import io.temporal.internal.nexus.OperationToken;
 import io.temporal.internal.nexus.OperationTokenType;
 import io.temporal.internal.nexus.OperationTokenUtil;
@@ -111,12 +110,10 @@ public void syncCompletedUpdateReturnsResult() {
 
   @Test
   public void asyncUpdateWorkflowOperationCompletes() {
-    // only run this test if the server supports update completion callbacks
-    // currently only for local server manual checks until the in-memory test
-    // server can support update completion callbacks
+    // Requires history.enableUpdateCallbacks and history.enableCHASMCallbacks
     assumeTrue(
         "server does not support update completion callbacks",
-        EnvironmentVariableUtils.readBooleanFlag("UPDATE_CALLBACKS_SUPPORTED"));
+        SDKTestWorkflowRule.useExternalService);
 
     TestWorkflows.TestWorkflow1 workflowStub =
         testWorkflowRule.newWorkflowStubTimeoutOptions(