Add support for sync update (#1749)

Add support for workflow update

Author: Quinn-With-Two-Ns

docker/buildkite/dynamicconfig/development.yaml

@@ -6,3 +6,6 @@ system.enableActivityEagerExecution:
 system.enableEagerWorkflowStart:
   - value: true
     constraints: {}
+frontend.enableUpdateWorkflowExecution:
+  - value: true
+    constraints: {}

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

@@ -0,0 +1,54 @@
+/*
+ * 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.client;
+
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.common.Experimental;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * UpdateHandle is a handle to an update workflow execution request that can be used to get the
+ * status of that update request.
+ */
+@Experimental
+public interface UpdateHandle<T> {
+  /**
+   * Gets the workflow execution this update request was sent to.
+   *
+   * @return the workflow execution this update was sent to.
+   */
+  WorkflowExecution getExecution();
+
+  /**
+   * Gets the unique ID of this update.
+   *
+   * @return the updates ID.
+   */
+  String getId();
+
+  /**
+   * Returns a {@link CompletableFuture} with the update workflow execution request result
+   * potentially waiting for the update to complete.
+   *
+   * @return future completed with the result of the update or an exception
+   */
+  CompletableFuture<T> getResultAsync();
+}

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

@@ -0,0 +1,54 @@
+/*
+ * 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.client;
+
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.common.Experimental;
+import java.util.concurrent.CompletableFuture;
+
+@Experimental
+final class UpdateHandleImpl<T> implements UpdateHandle<T> {
+
+  private final String id;
+  private final WorkflowExecution execution;
+  private final CompletableFuture<T> future;
+
+  UpdateHandleImpl(String id, WorkflowExecution execution, CompletableFuture<T> future) {
+    this.id = id;
+    this.execution = execution;
+    this.future = future;
+  }
+
+  @Override
+  public WorkflowExecution getExecution() {
+    return execution;
+  }
+
+  @Override
+  public String getId() {
+    return id;
+  }
+
+  @Override
+  public CompletableFuture<T> getResultAsync() {
+    return future;
+  }
+}

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

@@ -270,9 +270,11 @@ public void invoke(
       } else if (type == WorkflowMethodType.SIGNAL) {
         signalWorkflow(methodMetadata, untyped, method, args);
         result = null;
+      } else if (type == WorkflowMethodType.UPDATE) {
+        result = updateWorkflow(methodMetadata, untyped, method, args);
       } else {
         throw new IllegalArgumentException(
-            method + " is not annotated with @WorkflowMethod or @QueryMethod");
+            method + " is not annotated with @WorkflowMethod, @QueryMethod, @UpdateMethod");
       }
     }
 
@@ -306,6 +308,15 @@ private Object queryWorkflow(
       return untyped.query(queryType, method.getReturnType(), method.getGenericReturnType(), args);
     }
 
+    private Object updateWorkflow(
+        POJOWorkflowMethodMetadata methodMetadata,
+        WorkflowStub untyped,
+        Method method,
+        Object[] args) {
+      String updateType = methodMetadata.getName();
+      return untyped.update(updateType, method.getReturnType(), args);
+    }
+
     @SuppressWarnings("FutureReturnValueIgnored")
     private Object startWorkflow(WorkflowStub untyped, Method method, Object[] args) {
       WorkflowInvocationHandler.startWorkflow(untyped, args);

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

@@ -22,6 +22,7 @@
 
 import io.temporal.api.common.v1.WorkflowExecution;
 import io.temporal.api.enums.v1.QueryRejectCondition;
+import io.temporal.common.Experimental;
 import io.temporal.failure.CanceledFailure;
 import io.temporal.failure.TerminatedFailure;
 import io.temporal.failure.TimeoutFailure;
@@ -36,8 +37,8 @@
 
 /**
  * WorkflowStub is a client side stub to a single workflow instance. It can be used to start,
- * signal, query, wait for completion and cancel a workflow execution. Created through {@link
- * WorkflowClient#newUntypedWorkflowStub(String, WorkflowOptions)} or {@link
+ * signal, query, update, wait for completion and cancel a workflow execution. Created through
+ * {@link WorkflowClient#newUntypedWorkflowStub(String, WorkflowOptions)} or {@link
  * WorkflowClient#newUntypedWorkflowStub(WorkflowExecution, Optional)}.
  */
 public interface WorkflowStub {
@@ -73,6 +74,93 @@ static <T> WorkflowStub fromTyped(T typed) {
    */
   void signal(String signalName, Object... args);
 
+  /**
+   * Synchronously update a workflow execution by invoking its update handler. Usually a update
+   * handler is a method annotated with {@link io.temporal.workflow.UpdateMethod}.
+   *
+   * @param updateName name of the update handler. Usually it is a method name.
+   * @param resultClass class of the update return value
+   * @param <R> type of the update return value
+   * @param args update method arguments
+   * @return update result
+   * @throws WorkflowNotFoundException if the workflow execution doesn't exist or completed and
+   *     can't be signalled
+   * @throws WorkflowServiceException for all other failures including networking and service
+   *     availability issues
+   */
+  @Experimental
+  <R> R update(String updateName, Class<R> resultClass, Object... args);
+
+  /**
+   * Synchronously update a workflow execution by invoking its update handler. Usually a update
+   * handler is a method annotated with {@link io.temporal.workflow.UpdateMethod}.
+   *
+   * @param updateName name of the update handler. Usually it is a method name.
+   * @param updateId is an application-layer identifier for the requested update. It must be unique
+   *     within the scope of a workflow execution.
+   * @param firstExecutionRunId specifies the RunID expected to identify the first run in the
+   *     workflow execution chain. If this expectation does not match then the server will reject
+   *     the update request with an error.
+   * @param resultClass class of the update return value.
+   * @param <R> type of the update return value.
+   * @param resultType type of the workflow return value. Differs from resultClass for generic
+   *     types.
+   * @param args update method arguments
+   * @return update result
+   * @throws WorkflowNotFoundException if the workflow execution doesn't exist or completed and
+   *     can't be sent an update request
+   * @throws WorkflowServiceException for all other failures including networking and service
+   *     availability issues
+   */
+  @Experimental
+  <R> R update(
+      String updateName,
+      String updateId,
+      String firstExecutionRunId,
+      Class<R> resultClass,
+      Type resultType,
+      Object... args);
+
+  /**
+   * Asynchronously update a workflow execution by invoking its update handler and returning a
+   * handle to the update request. Usually a update handler is a method annotated with {@link
+   * io.temporal.workflow.UpdateMethod}.
+   *
+   * @param updateName name of the update handler. Usually it is a method name.
+   * @param resultClass class of the update return value
+   * @param <R> type of the update return value
+   * @param args update method arguments
+   * @return update reference that can be used to get the result of the update.
+   */
+  @Experimental
+  <R> UpdateHandle<R> startUpdate(String updateName, Class<R> resultClass, Object... args);
+
+  /**
+   * Asynchronously update a workflow execution by invoking its update handler and returning a
+   * handle to the update request.
+   *
+   * @param updateName name of the update handler. Usually it is a method name.
+   * @param updateId is an application-layer identifier for the requested update. It must be unique
+   *     within the scope of a workflow execution.
+   * @param firstExecutionRunId specifies the RunID expected to identify the first run in the
+   *     workflow execution chain. If this expectation does not match then the server will reject
+   *     the update request with an error.
+   * @param resultClass class of the update return value.
+   * @param <R> type of the update return value.
+   * @param resultType type of the workflow return value. Differs from resultClass for generic
+   *     types.
+   * @param args update method arguments
+   * @return update reference that can be used to get the result of the update.
+   */
+  @Experimental
+  <R> UpdateHandle<R> startUpdate(
+      String updateName,
+      String updateId,
+      String firstExecutionRunId,
+      Class<R> resultClass,
+      Type resultType,
+      Object... args);
+
   WorkflowExecution start(Object... args);
 
   WorkflowExecution signalWithStart(String signalName, Object[] signalArgs, Object[] startArgs);

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

@@ -29,6 +29,7 @@
 import io.temporal.api.errordetails.v1.WorkflowNotReadyFailure;
 import io.temporal.common.interceptors.Header;
 import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
+import io.temporal.common.interceptors.WorkflowClientCallsInterceptor.UpdateOutput;
 import io.temporal.failure.CanceledFailure;
 import io.temporal.serviceclient.CheckedExceptionWrapper;
 import io.temporal.serviceclient.StatusUtils;
@@ -292,6 +293,81 @@ public <R> R query(String queryType, Class<R> resultClass, Type resultType, Obje
     return result.getResult();
   }
 
+  @Override
+  public <R> R update(String updateName, Class<R> resultClass, Object... args) {
+    return update(updateName, "", "", resultClass, resultClass, args);
+  }
+
+  @Override
+  public <R> R update(
+      String updateName,
+      String updateId,
+      String firstExecutionRunId,
+      Class<R> resultClass,
+      Type resultType,
+      Object... args) {
+    checkStarted();
+    UpdateOutput<R> result;
+    WorkflowExecution targetExecution = execution.get();
+    try {
+      result =
+          workflowClientInvoker.update(
+              new WorkflowClientCallsInterceptor.UpdateInput<>(
+                  targetExecution,
+                  updateName,
+                  updateId,
+                  args,
+                  resultClass,
+                  resultType,
+                  firstExecutionRunId));
+      return result.getResult();
+
+    } catch (Exception e) {
+      Throwable throwable = throwAsWorkflowFailureException(e, targetExecution);
+      throw new WorkflowServiceException(targetExecution, workflowType.orElse(null), throwable);
+    }
+  }
+
+  @Override
+  public <R> UpdateHandle<R> startUpdate(String updateName, Class<R> resultClass, Object... args) {
+    return startUpdate(updateName, "", "", resultClass, resultClass, args);
+  }
+
+  @Override
+  public <R> UpdateHandle<R> startUpdate(
+      String updateName,
+      String updateId,
+      String firstExecutionRunId,
+      Class<R> resultClass,
+      Type resultType,
+      Object... args) {
+    checkStarted();
+    WorkflowExecution targetExecution = execution.get();
+
+    WorkflowClientCallsInterceptor.UpdateAsyncOutput<R> result =
+        workflowClientInvoker.updateAsync(
+            new WorkflowClientCallsInterceptor.UpdateInput<>(
+                targetExecution,
+                updateName,
+                updateId,
+                args,
+                resultClass,
+                resultType,
+                firstExecutionRunId));
+
+    return new UpdateHandleImpl<>(
+        updateId,
+        targetExecution,
+        result
+            .getResult()
+            .exceptionally(
+                e -> {
+                  Throwable throwable = throwAsWorkflowFailureException(e, targetExecution);
+                  throw new WorkflowServiceException(
+                      targetExecution, workflowType.orElse(null), throwable);
+                }));
+  }
+
   @Override
   public void cancel() {
     checkStarted();