Add initial support for StartWorkflowExecutionAsync (#884)

Add WorkflowStub#enqueueStart which uses the new StartWorkflowExecutionAsync to submit a workflow to be started at some point in the future.

Currently nearly all testing of the Cadence Java client is implemented as integration tests in WorkflowTests, running against either a locally running Cadence service or using the test environment. While this approach would be technically possible to configure (adding zookeeper and kafka to the docker-compose files), this would require reimplementing the integration tests already present on the Cadence server for all new features added. This approach doesn't scale well for adding new functionality to each client.

Instead, we add tests for this functionality via targeted unit tests at the different layers of the client. Add WorkflowClientInternalTest to use an in-memory fake server with stubbed responses to assert that we make exactly the correct request. This tests the entire Tchannel networking stack of the client end to end. Add WorkflowStubImplTest to test the specific semantics and state of the WorkflowStubImpl. State is particularly important with WorkflowStubs as they're intended to represent a single execution of a Workflow, so they normally capture the Workflow runId when starting a workflow. For StartWorkflowExecutionAsync we capture only the workflowId so that a stub can be used to enqueueStart(), then later qury or signal that workflow.

Additionally add tests for the thrift2proto mappers, including the MapperTestUtil assertions to ensure that all fields are present. This ensures that any update to the IDL files requires a corresponding change to either the test or mapper to account for the new field.

Author: natemort

src/main/java/com/uber/cadence/client/WorkflowStub.java

@@ -76,6 +76,13 @@ CompletableFuture<Void> signalAsyncWithTimeout(
   CompletableFuture<WorkflowExecution> startAsyncWithTimeout(
       long timeout, TimeUnit unit, Object... args);
 
+  WorkflowExecution enqueueStart(Object... args);
+
+  CompletableFuture<WorkflowExecution> enqueueStartAsync(Object... args);
+
+  CompletableFuture<WorkflowExecution> enqueueStartAsyncWithTimeout(
+      long timeout, TimeUnit unit, Object... args);
+
   WorkflowExecution signalWithStart(String signalName, Object[] signalArgs, Object[] startArgs);
 
   Optional<String> getWorkflowType();

src/main/java/com/uber/cadence/internal/compatibility/Thrift2ProtoAdapter.java

@@ -221,15 +221,25 @@ public StartWorkflowExecutionAsyncResponse StartWorkflowExecutionAsync(
       throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
           DomainNotActiveError, LimitExceededError, EntityNotExistsError,
           ClientVersionNotSupportedError, TException {
-    throw new IllegalArgumentException("unimplemented");
+    initializeStartWorkflowExecutionRequest(startRequest.getRequest());
+    try {
+      com.uber.cadence.api.v1.StartWorkflowExecutionAsyncResponse response =
+          grpcServiceStubs
+              .workflowBlockingStub()
+              .startWorkflowExecutionAsync(
+                  RequestMapper.startWorkflowExecutionAsyncRequest(startRequest));
+      return ResponseMapper.startWorkflowExecutionAsyncResponse(response);
+    } catch (StatusRuntimeException e) {
+      throw ErrorMapper.Error(e);
+    }
   }
 
   private StartWorkflowExecutionResponse startWorkflowExecution(
       StartWorkflowExecutionRequest startRequest)
       throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
           DomainNotActiveError, LimitExceededError, EntityNotExistsError,
           ClientVersionNotSupportedError, TException {
-    startRequest.setRequestId(UUID.randomUUID().toString());
+    initializeStartWorkflowExecutionRequest(startRequest);
     try {
       com.uber.cadence.api.v1.StartWorkflowExecutionResponse response =
           grpcServiceStubs
@@ -241,6 +251,10 @@ private StartWorkflowExecutionResponse startWorkflowExecution(
     }
   }
 
+  private void initializeStartWorkflowExecutionRequest(StartWorkflowExecutionRequest request) {
+    request.setRequestId(UUID.randomUUID().toString());
+  }
+
   @Override
   public GetWorkflowExecutionHistoryResponse GetWorkflowExecutionHistory(
       GetWorkflowExecutionHistoryRequest getRequest)
@@ -485,7 +499,7 @@ public StartWorkflowExecutionResponse SignalWithStartWorkflowExecution(
           LimitExceededError, WorkflowExecutionAlreadyStartedError, ClientVersionNotSupportedError,
           TException {
     try {
-      signalWithStartRequest.setRequestId(UUID.randomUUID().toString());
+      initializeSignalWithStartWorkflowExecution(signalWithStartRequest);
       com.uber.cadence.api.v1.SignalWithStartWorkflowExecutionResponse response =
           grpcServiceStubs
               .workflowBlockingStub()
@@ -503,7 +517,23 @@ public SignalWithStartWorkflowExecutionAsyncResponse SignalWithStartWorkflowExec
       throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
           DomainNotActiveError, LimitExceededError, EntityNotExistsError,
           ClientVersionNotSupportedError, TException {
-    throw new IllegalArgumentException("unimplemented");
+    try {
+      initializeSignalWithStartWorkflowExecution(signalWithStartRequest.getRequest());
+      com.uber.cadence.api.v1.SignalWithStartWorkflowExecutionAsyncResponse response =
+          grpcServiceStubs
+              .workflowBlockingStub()
+              .signalWithStartWorkflowExecutionAsync(
+                  RequestMapper.signalWithStartWorkflowExecutionAsyncRequest(
+                      signalWithStartRequest));
+      return ResponseMapper.signalWithStartWorkflowExecutionAsyncResponse(response);
+    } catch (StatusRuntimeException e) {
+      throw ErrorMapper.Error(e);
+    }
+  }
+
+  private void initializeSignalWithStartWorkflowExecution(
+      SignalWithStartWorkflowExecutionRequest request) {
+    request.setRequestId(UUID.randomUUID().toString());
   }
 
   @Override
@@ -825,7 +855,28 @@ public void StartWorkflowExecution(
   public void StartWorkflowExecutionAsync(
       StartWorkflowExecutionAsyncRequest startRequest, AsyncMethodCallback resultHandler)
       throws TException {
-    throw new IllegalArgumentException("unimplemented");
+    try {
+      initializeStartWorkflowExecutionRequest(startRequest.getRequest());
+      ListenableFuture<com.uber.cadence.api.v1.StartWorkflowExecutionAsyncResponse> resultFuture =
+          grpcServiceStubs
+              .workflowFutureStub()
+              .startWorkflowExecutionAsync(
+                  RequestMapper.startWorkflowExecutionAsyncRequest(startRequest));
+      resultFuture.addListener(
+          () -> {
+            try {
+              com.uber.cadence.api.v1.StartWorkflowExecutionAsyncResponse response =
+                  resultFuture.get();
+              resultHandler.onComplete(
+                  ResponseMapper.startWorkflowExecutionAsyncResponse(response));
+            } catch (Exception e) {
+              resultHandler.onError(e);
+            }
+          },
+          ForkJoinPool.commonPool());
+    } catch (StatusRuntimeException e) {
+      throw ErrorMapper.Error(e);
+    }
   }
 
   @Override
@@ -1124,7 +1175,7 @@ public void StartWorkflowExecutionWithTimeout(
       Long timeoutInMillis)
       throws TException {
     try {
-      startRequest.setRequestId(UUID.randomUUID().toString());
+      initializeStartWorkflowExecutionRequest(startRequest);
       ListenableFuture<com.uber.cadence.api.v1.StartWorkflowExecutionResponse> resultFuture =
           grpcServiceStubs
               .workflowFutureStub()
@@ -1145,6 +1196,37 @@ public void StartWorkflowExecutionWithTimeout(
     }
   }
 
+  @Override
+  public void StartWorkflowExecutionAsyncWithTimeout(
+      StartWorkflowExecutionAsyncRequest startAsyncRequest,
+      AsyncMethodCallback resultHandler,
+      Long timeoutInMillis)
+      throws TException {
+    try {
+      initializeStartWorkflowExecutionRequest(startAsyncRequest.getRequest());
+      ListenableFuture<com.uber.cadence.api.v1.StartWorkflowExecutionAsyncResponse> resultFuture =
+          grpcServiceStubs
+              .workflowFutureStub()
+              .withDeadline(Deadline.after(timeoutInMillis, TimeUnit.MILLISECONDS))
+              .startWorkflowExecutionAsync(
+                  RequestMapper.startWorkflowExecutionAsyncRequest(startAsyncRequest));
+      resultFuture.addListener(
+          () -> {
+            try {
+              com.uber.cadence.api.v1.StartWorkflowExecutionAsyncResponse response =
+                  resultFuture.get();
+              resultHandler.onComplete(
+                  ResponseMapper.startWorkflowExecutionAsyncResponse(response));
+            } catch (Exception e) {
+              resultHandler.onError(e);
+            }
+          },
+          ForkJoinPool.commonPool());
+    } catch (StatusRuntimeException e) {
+      throw ErrorMapper.Error(e);
+    }
+  }
+
   @Override
   public GetWorkflowExecutionHistoryResponse GetWorkflowExecutionHistoryWithTimeout(
       GetWorkflowExecutionHistoryRequest getRequest, Long timeoutInMillis) throws TException {

src/main/java/com/uber/cadence/internal/compatibility/proto/RequestMapper.java

@@ -84,8 +84,10 @@
 import com.uber.cadence.api.v1.RespondDecisionTaskFailedRequest;
 import com.uber.cadence.api.v1.RespondQueryTaskCompletedRequest;
 import com.uber.cadence.api.v1.ScanWorkflowExecutionsRequest;
+import com.uber.cadence.api.v1.SignalWithStartWorkflowExecutionAsyncRequest;
 import com.uber.cadence.api.v1.SignalWithStartWorkflowExecutionRequest;
 import com.uber.cadence.api.v1.SignalWorkflowExecutionRequest;
+import com.uber.cadence.api.v1.StartWorkflowExecutionAsyncRequest;
 import com.uber.cadence.api.v1.StartWorkflowExecutionRequest;
 import com.uber.cadence.api.v1.TerminateWorkflowExecutionRequest;
 import com.uber.cadence.api.v1.UpdateDomainRequest;
@@ -449,7 +451,7 @@ public static SignalWithStartWorkflowExecutionRequest signalWithStartWorkflowExe
     if (t.getDelayStartSeconds() > 0) {
       builder.setDelayStart(secondsToDuration(t.getDelayStartSeconds()));
     }
-    ;
+
     if (t.getIdentity() != null) {
       builder.setIdentity(t.getIdentity());
     }
@@ -464,6 +466,20 @@ public static SignalWithStartWorkflowExecutionRequest signalWithStartWorkflowExe
     return sb.build();
   }
 
+  public static SignalWithStartWorkflowExecutionAsyncRequest
+      signalWithStartWorkflowExecutionAsyncRequest(
+          com.uber.cadence.SignalWithStartWorkflowExecutionAsyncRequest t) {
+    if (t == null) {
+      return null;
+    }
+    SignalWithStartWorkflowExecutionAsyncRequest.Builder builder =
+        SignalWithStartWorkflowExecutionAsyncRequest.newBuilder();
+    if (t.getRequest() != null) {
+      builder.setRequest(signalWithStartWorkflowExecutionRequest(t.getRequest()));
+    }
+    return builder.build();
+  }
+
   public static SignalWorkflowExecutionRequest signalWorkflowExecutionRequest(
       com.uber.cadence.SignalWorkflowExecutionRequest t) {
     if (t == null) {
@@ -518,6 +534,19 @@ public static StartWorkflowExecutionRequest startWorkflowExecutionRequest(
     return request.build();
   }
 
+  public static StartWorkflowExecutionAsyncRequest startWorkflowExecutionAsyncRequest(
+      com.uber.cadence.StartWorkflowExecutionAsyncRequest t) {
+    if (t == null) {
+      return null;
+    }
+    StartWorkflowExecutionAsyncRequest.Builder builder =
+        StartWorkflowExecutionAsyncRequest.newBuilder();
+    if (t.getRequest() != null) {
+      builder.setRequest(startWorkflowExecutionRequest(t.getRequest()));
+    }
+    return builder.build();
+  }
+
   public static TerminateWorkflowExecutionRequest terminateWorkflowExecutionRequest(
       com.uber.cadence.TerminateWorkflowExecutionRequest t) {
     if (t == null) {

src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java

@@ -193,55 +193,58 @@ public void start(Listener<RespT> responseListener, Metadata headers) {
             span.finish();
           }
 
+          @SuppressWarnings("unchecked")
           @Override
           public void sendMessage(ReqT message) {
             if (Objects.equals(method.getBareMethodName(), "StartWorkflowExecution")
                 && message instanceof StartWorkflowExecutionRequest) {
               StartWorkflowExecutionRequest request = (StartWorkflowExecutionRequest) message;
-              Map<String, byte[]> headers = new HashMap<>();
-              tracingPropagator.inject(headers);
-              Header.Builder headerBuilder = request.getHeader().toBuilder();
-              headers.forEach(
-                  (k, v) -> {
-                    headerBuilder.putFields(
-                        k, Payload.newBuilder().setData(ByteString.copyFrom(v)).build());
-                  });
+              Header newHeader = addTracingHeaders(request.getHeader());
+
+              // cast should not throw error as we are using the builder
+              message = (ReqT) request.toBuilder().setHeader(newHeader).build();
+            } else if (Objects.equals(method.getBareMethodName(), "StartWorkflowExecutionAsync")
+                && message instanceof StartWorkflowExecutionAsyncRequest) {
+              StartWorkflowExecutionAsyncRequest request =
+                  (StartWorkflowExecutionAsyncRequest) message;
+              Header newHeader = addTracingHeaders(request.getRequest().getHeader());
 
               // cast should not throw error as we are using the builder
               message =
                   (ReqT)
-                      ((StartWorkflowExecutionRequest) message)
+                      request
                           .toBuilder()
-                          .setHeader(headerBuilder.build())
+                          .setRequest(request.getRequest().toBuilder().setHeader(newHeader))
                           .build();
-            }
-            if (Objects.equals(method.getBareMethodName(), "SignalWithStartWorkflowExecution")
+            } else if (Objects.equals(
+                    method.getBareMethodName(), "SignalWithStartWorkflowExecution")
                 && message instanceof SignalWithStartWorkflowExecutionRequest) {
               SignalWithStartWorkflowExecutionRequest request =
                   (SignalWithStartWorkflowExecutionRequest) message;
-              Map<String, byte[]> headers = new HashMap<>();
-              tracingPropagator.inject(headers);
-              Header.Builder headerBuilder = request.getStartRequest().getHeader().toBuilder();
-              headers.forEach(
-                  (k, v) -> {
-                    headerBuilder.putFields(
-                        k, Payload.newBuilder().setData(ByteString.copyFrom(v)).build());
-                  });
+              Header newHeader = addTracingHeaders(request.getStartRequest().getHeader());
 
               // cast should not throw error as we are using the builder
               message =
                   (ReqT)
-                      ((SignalWithStartWorkflowExecutionRequest) message)
+                      request
                           .toBuilder()
                           .setStartRequest(
-                              request
-                                  .getStartRequest()
-                                  .toBuilder()
-                                  .setHeader(headerBuilder.build()))
+                              request.getStartRequest().toBuilder().setHeader(newHeader))
                           .build();
             }
             super.sendMessage(message);
           }
+
+          private Header addTracingHeaders(Header header) {
+            Map<String, byte[]> headers = new HashMap<>();
+            tracingPropagator.inject(headers);
+            Header.Builder headerBuilder = header.toBuilder();
+            headers.forEach(
+                (k, v) ->
+                    headerBuilder.putFields(
+                        k, Payload.newBuilder().setData(ByteString.copyFrom(v)).build()));
+            return headerBuilder.build();
+          }
         };
       }
     };

src/main/java/com/uber/cadence/internal/compatibility/thrift/ResponseMapper.java

@@ -71,6 +71,8 @@
 import com.uber.cadence.RecordActivityTaskHeartbeatResponse;
 import com.uber.cadence.ResetWorkflowExecutionResponse;
 import com.uber.cadence.RespondDecisionTaskCompletedResponse;
+import com.uber.cadence.SignalWithStartWorkflowExecutionAsyncResponse;
+import com.uber.cadence.StartWorkflowExecutionAsyncResponse;
 import com.uber.cadence.StartWorkflowExecutionResponse;
 import com.uber.cadence.UpdateDomainResponse;
 import com.uber.cadence.api.v1.WorkflowQuery;
@@ -88,6 +90,11 @@ public static StartWorkflowExecutionResponse startWorkflowExecutionResponse(
     return startWorkflowExecutionResponse;
   }
 
+  public static StartWorkflowExecutionAsyncResponse startWorkflowExecutionAsyncResponse(
+      com.uber.cadence.api.v1.StartWorkflowExecutionAsyncResponse t) {
+    return t == null ? null : new StartWorkflowExecutionAsyncResponse();
+  }
+
   public static DescribeTaskListResponse describeTaskListResponse(
       com.uber.cadence.api.v1.DescribeTaskListResponse t) {
     if (t == null) {
@@ -399,6 +406,12 @@ public static StartWorkflowExecutionResponse signalWithStartWorkflowExecutionRes
     return startWorkflowExecutionResponse;
   }
 
+  public static SignalWithStartWorkflowExecutionAsyncResponse
+      signalWithStartWorkflowExecutionAsyncResponse(
+          com.uber.cadence.api.v1.SignalWithStartWorkflowExecutionAsyncResponse t) {
+    return t == null ? null : new SignalWithStartWorkflowExecutionAsyncResponse();
+  }
+
   public static UpdateDomainResponse updateDomainResponse(
       com.uber.cadence.api.v1.UpdateDomainResponse t) {
     if (t == null) {

src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternal.java

@@ -39,6 +39,12 @@ CompletableFuture<WorkflowExecution> startWorkflowAsync(
   CompletableFuture<WorkflowExecution> startWorkflowAsync(
       StartWorkflowExecutionParameters startParameters, Long timeoutInMillis);
 
+  void enqueueStartWorkflow(StartWorkflowExecutionParameters startParameters)
+      throws WorkflowExecutionAlreadyStartedError;
+
+  CompletableFuture<Void> enqueueStartWorkflowAsync(
+      StartWorkflowExecutionParameters startParameters, Long timeoutInMillis);
+
   void signalWorkflowExecution(SignalExternalWorkflowParameters signalParameters);
 
   CompletableFuture<Void> signalWorkflowExecutionAsync(