Use the same Request ID for retries (#915)

Currently the RequestId is initialized in the implementation of IWorkflowService (either Thrift2ProtoAdapter or WorkflowServiceTChannel). Since retries are handled within GenericWorkflowClientExternalImpl, each retry uses a different request ID. This causes Cadence's idempotency feature not to work correctly for retries from the Java client.

Move Request ID initialization to GenericWorkflowClientExternalImpl, and for backwards compatibility (since users may be interacting with the IWorkflowService directly) continue setting the request ID if not specified.

Author: natemort

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

@@ -252,7 +252,9 @@ private StartWorkflowExecutionResponse startWorkflowExecution(
   }
 
   private void initializeStartWorkflowExecutionRequest(StartWorkflowExecutionRequest request) {
-    request.setRequestId(UUID.randomUUID().toString());
+    if (!request.isSetRequestId()) {
+      request.setRequestId(UUID.randomUUID().toString());
+    }
   }
 
   @Override
@@ -466,7 +468,9 @@ public void RequestCancelWorkflowExecution(RequestCancelWorkflowExecutionRequest
       throws BadRequestError, EntityNotExistsError, CancellationAlreadyRequestedError,
           ServiceBusyError, DomainNotActiveError, LimitExceededError,
           ClientVersionNotSupportedError, WorkflowExecutionAlreadyCompletedError, TException {
-    cancelRequest.setRequestId(UUID.randomUUID().toString());
+    if (!cancelRequest.isSetRequestId()) {
+      cancelRequest.setRequestId(UUID.randomUUID().toString());
+    }
     try {
       grpcServiceStubs
           .workflowBlockingStub()
@@ -482,7 +486,9 @@ public void SignalWorkflowExecution(SignalWorkflowExecutionRequest signalRequest
       throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError,
           LimitExceededError, ClientVersionNotSupportedError,
           WorkflowExecutionAlreadyCompletedError, TException {
-    signalRequest.setRequestId(UUID.randomUUID().toString());
+    if (!signalRequest.isSetRequestId()) {
+      signalRequest.setRequestId(UUID.randomUUID().toString());
+    }
     try {
       grpcServiceStubs
           .workflowBlockingStub()
@@ -533,7 +539,9 @@ public SignalWithStartWorkflowExecutionAsyncResponse SignalWithStartWorkflowExec
 
   private void initializeSignalWithStartWorkflowExecution(
       SignalWithStartWorkflowExecutionRequest request) {
-    request.setRequestId(UUID.randomUUID().toString());
+    if (!request.isSetRequestId()) {
+      request.setRequestId(UUID.randomUUID().toString());
+    }
   }
 
   @Override
@@ -542,7 +550,9 @@ public ResetWorkflowExecutionResponse ResetWorkflowExecution(
       throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError,
           LimitExceededError, ClientVersionNotSupportedError, TException {
     try {
-      resetRequest.setRequestId(UUID.randomUUID().toString());
+      if (!resetRequest.isSetRequestId()) {
+        resetRequest.setRequestId(UUID.randomUUID().toString());
+      }
       com.uber.cadence.api.v1.ResetWorkflowExecutionResponse response =
           grpcServiceStubs
               .workflowBlockingStub()
@@ -980,7 +990,9 @@ public void SignalWorkflowExecution(
       SignalWorkflowExecutionRequest signalRequest, AsyncMethodCallback resultHandler)
       throws TException {
     try {
-      signalRequest.setRequestId(UUID.randomUUID().toString());
+      if (!signalRequest.isSetRequestId()) {
+        signalRequest.setRequestId(UUID.randomUUID().toString());
+      }
       ListenableFuture<com.uber.cadence.api.v1.SignalWorkflowExecutionResponse> resultFuture =
           grpcServiceStubs
               .workflowFutureStub()

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

@@ -244,7 +244,8 @@ public void onError(Exception exception) {
 
   private StartWorkflowExecutionRequest getStartRequest(
       StartWorkflowExecutionParameters startParameters) {
-    StartWorkflowExecutionRequest request = new StartWorkflowExecutionRequest();
+    StartWorkflowExecutionRequest request =
+        new StartWorkflowExecutionRequest().setRequestId(generateUniqueId());
     request.setDomain(domain);
     if (startParameters.getInput() != null) {
       request.setInput(startParameters.getInput());
@@ -383,7 +384,8 @@ public void onError(Exception exception) {
 
   private SignalWorkflowExecutionRequest getSignalRequest(
       SignalExternalWorkflowParameters signalParameters) {
-    SignalWorkflowExecutionRequest request = new SignalWorkflowExecutionRequest();
+    SignalWorkflowExecutionRequest request =
+        new SignalWorkflowExecutionRequest().setRequestId(generateUniqueId());
     request.setDomain(domain);
     request.setInput(signalParameters.getInput());
     request.setSignalName(signalParameters.getSignalName());
@@ -466,7 +468,8 @@ private WorkflowExecution signalWithStartWorkflowInternal(
 
   private SignalWithStartWorkflowExecutionRequest createSignalWithStartRequest(
       SignalWithStartWorkflowExecutionParameters parameters) {
-    SignalWithStartWorkflowExecutionRequest request = new SignalWithStartWorkflowExecutionRequest();
+    SignalWithStartWorkflowExecutionRequest request =
+        new SignalWithStartWorkflowExecutionRequest().setRequestId(generateUniqueId());
     request.setDomain(domain);
     StartWorkflowExecutionParameters startParameters = parameters.getStartParameters();
     request.setSignalName(parameters.getSignalName());
@@ -509,7 +512,8 @@ private SignalWithStartWorkflowExecutionRequest createSignalWithStartRequest(
 
   @Override
   public void requestCancelWorkflowExecution(WorkflowExecution execution) {
-    RequestCancelWorkflowExecutionRequest request = new RequestCancelWorkflowExecutionRequest();
+    RequestCancelWorkflowExecutionRequest request =
+        new RequestCancelWorkflowExecutionRequest().setRequestId(generateUniqueId());
     request.setDomain(domain);
     request.setWorkflowExecution(execution);
     try {
@@ -544,8 +548,7 @@ public QueryWorkflowResponse queryWorkflow(QueryWorkflowParameters queryParamete
 
   @Override
   public String generateUniqueId() {
-    String workflowId = UUID.randomUUID().toString();
-    return workflowId;
+    return UUID.randomUUID().toString();
   }
 
   @Override

src/main/java/com/uber/cadence/serviceclient/WorkflowServiceTChannel.java

@@ -712,7 +712,9 @@ private StartWorkflowExecutionAsyncResponse startWorkflowExecutionAsync(
   }
 
   private void initializeStartWorkflowRequest(StartWorkflowExecutionRequest startRequest) {
-    startRequest.setRequestId(UUID.randomUUID().toString());
+    if (!startRequest.isSetRequestId()) {
+      startRequest.setRequestId(UUID.randomUUID().toString());
+    }
     // Write span context to header
     if (!startRequest.isSetHeader()) {
       startRequest.setHeader(new Header());
@@ -1412,7 +1414,9 @@ public void RequestCancelWorkflowExecution(RequestCancelWorkflowExecutionRequest
 
   private void requestCancelWorkflowExecution(RequestCancelWorkflowExecutionRequest cancelRequest)
       throws TException {
-    cancelRequest.setRequestId(UUID.randomUUID().toString());
+    if (!cancelRequest.isSetRequestId()) {
+      cancelRequest.setRequestId(UUID.randomUUID().toString());
+    }
     ThriftResponse<WorkflowService.RequestCancelWorkflowExecution_result> response = null;
     try {
       ThriftRequest<WorkflowService.RequestCancelWorkflowExecution_args> request =
@@ -1667,7 +1671,9 @@ private StartWorkflowExecutionResponse signalWithStartWorkflowExecution(
 
   private void initializeSignalWithStartWorkflowRequest(
       SignalWithStartWorkflowExecutionRequest request) {
-    request.setRequestId(UUID.randomUUID().toString());
+    if (!request.isSetRequestId()) {
+      request.setRequestId(UUID.randomUUID().toString());
+    }
     // Write span context to header
     if (!request.isSetHeader()) {
       request.setHeader(new Header());

src/test/java/com/uber/cadence/FakeWorkflowServiceRule.java

@@ -10,8 +10,10 @@
 import com.uber.tchannel.messages.ThriftResponse;
 import io.opentracing.mock.MockTracer;
 import java.util.Map;
+import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import org.junit.rules.ExternalResource;
 
 /**
@@ -21,7 +23,7 @@
  */
 public class FakeWorkflowServiceRule extends ExternalResource {
 
-  private final Map<String, StubbedResponse<?>> stubbedResponses = new ConcurrentHashMap<>();
+  private final Map<String, StubbedEndpoint> stubbedEndpoints = new ConcurrentHashMap<>();
   private final MockTracer tracer = new MockTracer();
   private TChannel tChannel;
   private IWorkflowService clientConn;
@@ -33,11 +35,16 @@ protected void before() throws Throwable {
         new ThriftRequestHandler<Object, Object>() {
           @Override
           public ThriftResponse<Object> handleImpl(ThriftRequest<Object> request) {
+            StubbedEndpoint endpoint = stubbedEndpoints.get(request.getEndpoint());
+            if (endpoint == null) {
+              throw new IllegalStateException(
+                  "Endpoint " + request.getEndpoint() + " was not stubbed");
+            }
             @SuppressWarnings("rawtypes")
-            StubbedResponse stub = stubbedResponses.get(request.getEndpoint());
+            StubbedResponse stub = endpoint.getNext();
             if (stub == null) {
               throw new IllegalStateException(
-                  "Endpoint " + request.getEndpoint() + " was not stubbed");
+                  "Exhausted all invocations of  " + request.getEndpoint());
             }
             //noinspection unchecked
             stub.future.complete(request.getBody(stub.requestType));
@@ -59,7 +66,7 @@ public ThriftResponse<Object> handleImpl(ThriftRequest<Object> request) {
 
   @Override
   protected void after() {
-    stubbedResponses.clear();
+    stubbedEndpoints.clear();
     if (clientConn != null) {
       clientConn.close();
     }
@@ -71,7 +78,7 @@ protected void after() {
 
   public void resetStubs() {
     tracer.reset();
-    stubbedResponses.clear();
+    stubbedEndpoints.clear();
   }
 
   public IWorkflowService getClient() {
@@ -82,19 +89,37 @@ public MockTracer getTracer() {
     return tracer;
   }
 
-  public <V> CompletableFuture<V> stubEndpoint(
+  public <V> CompletableFuture<V> stubSuccess(
       String endpoint, Class<V> requestType, Object response) {
+    return stubEndpoint(endpoint, requestType, ResponseCode.OK, response);
+  }
+
+  public <V> CompletableFuture<V> stubError(
+      String endpoint, Class<V> requestType, Object response) {
+    return stubEndpoint(endpoint, requestType, ResponseCode.Error, response);
+  }
+
+  public <V> CompletableFuture<V> stubEndpoint(
+      String endpoint, Class<V> requestType, ResponseCode code, Object response) {
     CompletableFuture<V> future = new CompletableFuture<>();
-    StubbedResponse<?> existingStub =
-        stubbedResponses.putIfAbsent(
-            endpoint, new StubbedResponse<>(response, ResponseCode.OK, future, requestType));
-    if (existingStub != null) {
-      throw new IllegalStateException(
-          "Endpoint " + endpoint + " was already stubbed to return " + existingStub.body);
-    }
+    StubbedEndpoint endpointStub =
+        stubbedEndpoints.computeIfAbsent(endpoint, id -> new StubbedEndpoint());
+    endpointStub.addStub(new StubbedResponse<>(response, code, future, requestType));
     return future;
   }
 
+  private static class StubbedEndpoint {
+    private final Queue<StubbedResponse<?>> responses = new ConcurrentLinkedQueue<>();
+
+    public void addStub(StubbedResponse<?> response) {
+      responses.add(response);
+    }
+
+    public StubbedResponse<?> getNext() {
+      return responses.poll();
+    }
+  }
+
   private static class StubbedResponse<V> {
     private final Object body;
     private final ResponseCode code;