Add support for SignalWithStartWorkflowExecutionAsync (#887)

Add support to schedule a SignalWithStartWorkflowExecution operation to be performed at a future date via SignalWithStartWorkflowExecutionAsync. This is exposed in the java client in two ways:
- WorkflowStub#enqueueSignalWithStart, which presents an untyped interface.
- WorkflowClient#enqueueSignalWithStart, with presents a statically typed interface.

The existing mechanisms for SignalWithStart don't currently have an async equivalent so we don't introduce one here.

Additionally add full test coverage for tracing information being injected into headers both in GRPC and TChannel.

Author: natemort

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

@@ -227,7 +227,8 @@ WorkflowStub newUntypedWorkflowStub(
    * running. The batch before invocation must contain exactly two operations. One annotated
    * with @WorkflowMethod and another with @SignalMethod.
    *
-   * @return batch request used to call {@link #signalWithStart(BatchRequest)}
+   * @return batch request used to call {@link #signalWithStart(BatchRequest)} or {@link
+   *     #enqueueSignalWithStart(BatchRequest)}
    */
   BatchRequest newSignalWithStartRequest();
 
@@ -239,6 +240,20 @@ WorkflowStub newUntypedWorkflowStub(
    */
   WorkflowExecution signalWithStart(BatchRequest signalWithStartBatch);
 
+  /**
+   * Schedules a SignalWithStart operation to be performed at a future date via
+   * SignalWithStartWorkflowExecutionAsync. This requires that async execution has been enabled for
+   * this domain.
+   *
+   * <p>Note that the returned WorkflowExecution will <b>NOT</b> contain a {@code runId}, only a
+   * {@code workflowId}. This is because the {@code runId} is only determined at the time the
+   * workflow actually starts.
+   *
+   * @param signalWithStartBatch Must be created with {@link #newSignalWithStartRequest()}
+   * @return WorkflowExecution containing only the workflowId
+   */
+  WorkflowExecution enqueueSignalWithStart(BatchRequest signalWithStartBatch);
+
   /**
    * Refreshes all the tasks of a given workflow.
    *

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

@@ -85,6 +85,9 @@ CompletableFuture<WorkflowExecution> enqueueStartAsyncWithTimeout(
 
   WorkflowExecution signalWithStart(String signalName, Object[] signalArgs, Object[] startArgs);
 
+  WorkflowExecution enqueueSignalWithStart(
+      String signalName, Object[] signalArgs, Object[] startArgs);
+
   Optional<String> getWorkflowType();
 
   WorkflowExecution getExecution();

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

@@ -231,6 +231,30 @@ public void sendMessage(ReqT message) {
                           .setStartRequest(
                               request.getStartRequest().toBuilder().setHeader(newHeader))
                           .build();
+            } else if (Objects.equals(
+                    method.getBareMethodName(), "SignalWithStartWorkflowExecutionAsync")
+                && message instanceof SignalWithStartWorkflowExecutionAsyncRequest) {
+              SignalWithStartWorkflowExecutionAsyncRequest request =
+                  (SignalWithStartWorkflowExecutionAsyncRequest) message;
+              Header newHeader =
+                  addTracingHeaders(request.getRequest().getStartRequest().getHeader());
+
+              // cast should not throw error as we are using the builder
+              message =
+                  (ReqT)
+                      request
+                          .toBuilder()
+                          .setRequest(
+                              request
+                                  .getRequest()
+                                  .toBuilder()
+                                  .setStartRequest(
+                                      request
+                                          .getRequest()
+                                          .getStartRequest()
+                                          .toBuilder()
+                                          .setHeader(newHeader)))
+                          .build();
             }
             super.sendMessage(message);
           }

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

@@ -56,6 +56,9 @@ CompletableFuture<Void> signalWorkflowExecutionAsync(
   WorkflowExecution signalWithStartWorkflowExecution(
       SignalWithStartWorkflowExecutionParameters parameters);
 
+  WorkflowExecution enqueueSignalWithStartWorkflowExecution(
+      SignalWithStartWorkflowExecutionParameters parameters);
+
   void requestCancelWorkflowExecution(WorkflowExecution execution);
 
   QueryWorkflowResponse queryWorkflow(QueryWorkflowParameters queryParameters);

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

@@ -25,6 +25,7 @@
 import com.uber.cadence.RequestCancelWorkflowExecutionRequest;
 import com.uber.cadence.RetryPolicy;
 import com.uber.cadence.SearchAttributes;
+import com.uber.cadence.SignalWithStartWorkflowExecutionAsyncRequest;
 import com.uber.cadence.SignalWithStartWorkflowExecutionRequest;
 import com.uber.cadence.SignalWorkflowExecutionRequest;
 import com.uber.cadence.StartWorkflowExecutionAsyncRequest;
@@ -411,8 +412,60 @@ public WorkflowExecution signalWithStartWorkflowExecution(
     }
   }
 
+  @Override
+  public WorkflowExecution enqueueSignalWithStartWorkflowExecution(
+      SignalWithStartWorkflowExecutionParameters parameters) {
+    try {
+      return enqueueSignalWithStartWorkflowInternal(parameters);
+    } finally {
+      Map<String, String> tags =
+          new ImmutableMap.Builder<String, String>(3)
+              .put(
+                  MetricsTag.WORKFLOW_TYPE,
+                  parameters.getStartParameters().getWorkflowType().getName())
+              .put(MetricsTag.TASK_LIST, parameters.getStartParameters().getTaskList())
+              .put(MetricsTag.DOMAIN, domain)
+              .build();
+      metricsScope
+          .tagged(tags)
+          .counter(MetricsType.WORKFLOW_SIGNAL_WITH_START_ASYNC_COUNTER)
+          .inc(1);
+    }
+  }
+
+  private WorkflowExecution enqueueSignalWithStartWorkflowInternal(
+      SignalWithStartWorkflowExecutionParameters parameters) {
+    SignalWithStartWorkflowExecutionAsyncRequest request =
+        new SignalWithStartWorkflowExecutionAsyncRequest()
+            .setRequest(createSignalWithStartRequest(parameters));
+    try {
+      RpcRetryer.retryWithResult(
+          RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS,
+          () -> service.SignalWithStartWorkflowExecutionAsync(request));
+      return new WorkflowExecution().setWorkflowId(request.getRequest().getWorkflowId());
+    } catch (TException e) {
+      throw CheckedExceptionWrapper.wrap(e);
+    }
+  }
+
   private WorkflowExecution signalWithStartWorkflowInternal(
       SignalWithStartWorkflowExecutionParameters parameters) {
+    SignalWithStartWorkflowExecutionRequest request = createSignalWithStartRequest(parameters);
+    try {
+      StartWorkflowExecutionResponse result =
+          RpcRetryer.retryWithResult(
+              RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS,
+              () -> service.SignalWithStartWorkflowExecution(request));
+      return new WorkflowExecution()
+          .setRunId(result.getRunId())
+          .setWorkflowId(request.getWorkflowId());
+    } catch (TException e) {
+      throw CheckedExceptionWrapper.wrap(e);
+    }
+  }
+
+  private SignalWithStartWorkflowExecutionRequest createSignalWithStartRequest(
+      SignalWithStartWorkflowExecutionParameters parameters) {
     SignalWithStartWorkflowExecutionRequest request = new SignalWithStartWorkflowExecutionRequest();
     request.setDomain(domain);
     StartWorkflowExecutionParameters startParameters = parameters.getStartParameters();
@@ -451,19 +504,7 @@ private WorkflowExecution signalWithStartWorkflowInternal(
     if (startParameters.getDelayStart() != null) {
       request.setDelayStartSeconds((int) startParameters.getDelayStart().getSeconds());
     }
-    StartWorkflowExecutionResponse result;
-    try {
-      result =
-          RpcRetryer.retryWithResult(
-              RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS,
-              () -> service.SignalWithStartWorkflowExecution(request));
-    } catch (TException e) {
-      throw CheckedExceptionWrapper.wrap(e);
-    }
-    WorkflowExecution execution = new WorkflowExecution();
-    execution.setRunId(result.getRunId());
-    execution.setWorkflowId(request.getWorkflowId());
-    return execution;
+    return request;
   }
 
   @Override

src/main/java/com/uber/cadence/internal/metrics/MetricsType.java

@@ -43,6 +43,8 @@ public class MetricsType {
       CADENCE_METRICS_PREFIX + "workflow-get-history-latency";
   public static final String WORKFLOW_SIGNAL_WITH_START_COUNTER =
       CADENCE_METRICS_PREFIX + "workflow-signal-with-start";
+  public static final String WORKFLOW_SIGNAL_WITH_START_ASYNC_COUNTER =
+      CADENCE_METRICS_PREFIX + "workflow-signal-with-start-async";
 
   public static final String DECISION_POLL_COUNTER = CADENCE_METRICS_PREFIX + "decision-poll-total";
   public static final String DECISION_POLL_FAILED_COUNTER =

src/main/java/com/uber/cadence/internal/sync/SignalWithStartBatchRequest.java

@@ -24,6 +24,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 
 final class SignalWithStartBatchRequest implements BatchRequest {
 
@@ -33,9 +34,17 @@ final class SignalWithStartBatchRequest implements BatchRequest {
   private String signalName;
   private Object[] signalArgs;
   private Object[] startArgs;
-  private AtomicBoolean invoked = new AtomicBoolean();
+  private final AtomicBoolean invoked = new AtomicBoolean();
 
-  WorkflowExecution invoke() {
+  WorkflowExecution enqueue() {
+    return invoke(this::enqueueSignalWithStart);
+  }
+
+  WorkflowExecution execute() {
+    return invoke(this::signalWithStart);
+  }
+
+  private WorkflowExecution invoke(Supplier<WorkflowExecution> commitFn) {
     if (!invoked.compareAndSet(false, true)) {
       throw new IllegalStateException(
           "A batch instance can be used only for a single signalWithStart call");
@@ -46,7 +55,7 @@ WorkflowExecution invoke() {
       for (Functions.Proc request : requests) {
         request.apply();
       }
-      return signalWithStart();
+      return commitFn.get();
     } finally {
       WorkflowInvocationHandler.closeAsyncInvocation();
     }
@@ -56,6 +65,10 @@ private WorkflowExecution signalWithStart() {
     return stub.signalWithStart(signalName, signalArgs, startArgs);
   }
 
+  private WorkflowExecution enqueueSignalWithStart() {
+    return stub.enqueueSignalWithStart(signalName, signalArgs, startArgs);
+  }
+
   void signal(WorkflowStub stub, String signalName, Object[] args) {
     setStub(stub);
     this.signalName = signalName;

src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java

@@ -1012,6 +1012,12 @@ public WorkflowExecution signalWithStart(
         return next.signalWithStart(signalName, signalArgs, startArgs);
       }
 
+      @Override
+      public WorkflowExecution enqueueSignalWithStart(
+          String signalName, Object[] signalArgs, Object[] startArgs) {
+        return next.enqueueSignalWithStart(signalName, signalArgs, startArgs);
+      }
+
       @Override
       public Optional<String> getWorkflowType() {
         return next.getWorkflowType();

src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java

@@ -210,7 +210,12 @@ public BatchRequest newSignalWithStartRequest() {
 
   @Override
   public WorkflowExecution signalWithStart(BatchRequest signalWithStartBatch) {
-    return ((SignalWithStartBatchRequest) signalWithStartBatch).invoke();
+    return ((SignalWithStartBatchRequest) signalWithStartBatch).execute();
+  }
+
+  @Override
+  public WorkflowExecution enqueueSignalWithStart(BatchRequest signalWithStartBatch) {
+    return ((SignalWithStartBatchRequest) signalWithStartBatch).enqueue();
   }
 
   @Override

src/main/java/com/uber/cadence/internal/sync/WorkflowStubImpl.java

@@ -323,6 +323,31 @@ public WorkflowExecution signalWithStart(
         WorkflowOptions.merge(null, null, null, options.get()), signalName, signalArgs, startArgs);
   }
 
+  @Override
+  public WorkflowExecution enqueueSignalWithStart(
+      String signalName, Object[] signalArgs, Object[] startArgs) {
+    if (!options.isPresent()) {
+      throw new IllegalStateException("Required parameter WorkflowOptions is missing");
+    }
+    return enqueueSignalWithStart(
+        WorkflowOptions.merge(null, null, null, options.get()), signalName, signalArgs, startArgs);
+  }
+
+  private WorkflowExecution enqueueSignalWithStart(
+      WorkflowOptions options, String signalName, Object[] signalArgs, Object[] startArgs) {
+    StartWorkflowExecutionParameters sp = getStartWorkflowExecutionParameters(options, startArgs);
+
+    byte[] signalInput = dataConverter.toData(signalArgs);
+    SignalWithStartWorkflowExecutionParameters p =
+        new SignalWithStartWorkflowExecutionParameters(sp, signalName, signalInput);
+    try {
+      execution.set(genericClient.enqueueSignalWithStartWorkflowExecution(p));
+    } catch (Exception e) {
+      throw new WorkflowServiceException(execution.get(), workflowType, e);
+    }
+    return execution.get();
+  }
+
   @Override
   public Optional<String> getWorkflowType() {
     return workflowType;