Added support for overriding task list when calling continue as new (#223)

Author: mfateev

src/main/java/com/uber/cadence/internal/replay/ContinueAsNewWorkflowExecutionParameters.java

@@ -17,7 +17,6 @@
 
 package com.uber.cadence.internal.replay;
 
-import com.uber.cadence.ChildPolicy;
 import java.nio.charset.StandardCharsets;
 
 public final class ContinueAsNewWorkflowExecutionParameters {
@@ -26,7 +25,6 @@ public final class ContinueAsNewWorkflowExecutionParameters {
   private byte[] input;
   private String taskList;
   private int taskStartToCloseTimeoutSeconds;
-  private ChildPolicy childPolicy;
   private String workflowType;
 
   public void setWorkflowType(String workflowType) {
@@ -85,19 +83,6 @@ public void setTaskStartToCloseTimeoutSeconds(int taskStartToCloseTimeoutSeconds
     this.taskStartToCloseTimeoutSeconds = taskStartToCloseTimeoutSeconds;
   }
 
-  public ChildPolicy getChildPolicy() {
-    return childPolicy;
-  }
-
-  public void setChildPolicy(ChildPolicy childPolicy) {
-    this.childPolicy = childPolicy;
-  }
-
-  public ContinueAsNewWorkflowExecutionParameters withChildPolicy(ChildPolicy childPolicy) {
-    this.childPolicy = childPolicy;
-    return this;
-  }
-
   public ContinueAsNewWorkflowExecutionParameters withTaskStartToCloseTimeoutSeconds(
       int taskStartToCloseTimeoutSeconds) {
     this.taskStartToCloseTimeoutSeconds = taskStartToCloseTimeoutSeconds;

src/main/java/com/uber/cadence/internal/replay/DecisionContext.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.replay;
 
+import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.converter.DataConverter;
@@ -68,6 +69,8 @@ public interface DecisionContext extends ReplayAware {
 
   Duration getDecisionTaskTimeout();
 
+  ChildPolicy getChildPolicy();
+
   /**
    * Used to dynamically schedule an activity for execution
    *

src/main/java/com/uber/cadence/internal/replay/DecisionContextImpl.java

@@ -17,12 +17,7 @@
 
 package com.uber.cadence.internal.replay;
 
-import com.uber.cadence.HistoryEvent;
-import com.uber.cadence.PollForDecisionTaskResponse;
-import com.uber.cadence.TimerFiredEventAttributes;
-import com.uber.cadence.WorkflowExecution;
-import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
-import com.uber.cadence.WorkflowType;
+import com.uber.cadence.*;
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.internal.metrics.ReplayAwareScope;
 import com.uber.cadence.workflow.Functions.Func;
@@ -128,6 +123,11 @@ public Duration getDecisionTaskTimeout() {
     return Duration.ofSeconds(workflowContext.getDecisionTaskTimeoutSeconds());
   }
 
+  @Override
+  public ChildPolicy getChildPolicy() {
+    return workflowContext.getChildPolicy();
+  }
+
   @Override
   public String getTaskList() {
     return workflowContext.getTaskList();

src/main/java/com/uber/cadence/internal/replay/WorkflowContext.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.replay;
 
+import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.PollForDecisionTaskResponse;
 import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
 
@@ -116,4 +117,8 @@ String getDomain() {
   private WorkflowExecutionStartedEventAttributes getWorkflowStartedEventAttributes() {
     return startedAttributes;
   }
+
+  public ChildPolicy getChildPolicy() {
+    return startedAttributes.getChildPolicy();
+  }
 }

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

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.sync;
 
+import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.converter.DataConverter;
@@ -528,6 +529,11 @@ public Duration getDecisionTaskTimeout() {
       throw new UnsupportedOperationException("not implemented");
     }
 
+    @Override
+    public ChildPolicy getChildPolicy() {
+      return ChildPolicy.TERMINATE;
+    }
+
     @Override
     public Consumer<Exception> scheduleActivityTask(
         ExecuteActivityParameters parameters, BiConsumer<byte[], Exception> callback) {

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

@@ -526,10 +526,12 @@ public void continueAsNew(
       parameters.setWorkflowType(workflowType.get());
     }
     if (options.isPresent()) {
+      ContinueAsNewOptions ops = options.get();
       parameters.setExecutionStartToCloseTimeoutSeconds(
-          (int) options.get().getExecutionStartToCloseTimeout().getSeconds());
+          (int) ops.getExecutionStartToCloseTimeout().getSeconds());
       parameters.setTaskStartToCloseTimeoutSeconds(
-          (int) options.get().getTaskStartToCloseTimeout().getSeconds());
+          (int) ops.getTaskStartToCloseTimeout().getSeconds());
+      parameters.setTaskList(ops.getTaskList());
     }
     parameters.setInput(getDataConverter().toData(args));
     context.continueAsNewOnCompletion(parameters);

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

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.sync;
 
+import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.internal.replay.DecisionContext;
 import com.uber.cadence.workflow.WorkflowInfo;
 import java.time.Duration;
@@ -58,4 +59,9 @@ public String getTaskList() {
   public Duration getExecutionStartToCloseTimeout() {
     return context.getExecutionStartToCloseTimeout();
   }
+
+  @Override
+  public ChildPolicy getChildPolicy() {
+    return context.getChildPolicy();
+  }
 }

src/main/java/com/uber/cadence/workflow/ContinueAsNewOptions.java

@@ -17,7 +17,6 @@
 
 package com.uber.cadence.workflow;
 
-import com.uber.cadence.ChildPolicy;
 import com.uber.cadence.internal.common.OptionsUtils;
 import java.time.Duration;
 
@@ -28,7 +27,6 @@ public static final class Builder {
     private Duration executionStartToCloseTimeout;
     private String taskList;
     private Duration taskStartToCloseTimeout;
-    private ChildPolicy childPolicy;
 
     public Builder setExecutionStartToCloseTimeout(Duration executionStartToCloseTimeout) {
       this.executionStartToCloseTimeout = executionStartToCloseTimeout;
@@ -45,34 +43,23 @@ public Builder setTaskStartToCloseTimeout(Duration taskStartToCloseTimeout) {
       return this;
     }
 
-    public Builder setChildPolicy(ChildPolicy childPolicy) {
-      this.childPolicy = childPolicy;
-      return this;
-    }
-
     public ContinueAsNewOptions build() {
       return new ContinueAsNewOptions(
           OptionsUtils.roundUpToSeconds(executionStartToCloseTimeout),
           taskList,
-          OptionsUtils.roundUpToSeconds(taskStartToCloseTimeout),
-          childPolicy);
+          OptionsUtils.roundUpToSeconds(taskStartToCloseTimeout));
     }
   }
 
   private final Duration executionStartToCloseTimeout;
   private final String taskList;
   private final Duration taskStartToCloseTimeout;
-  private final ChildPolicy childPolicy;
 
   public ContinueAsNewOptions(
-      Duration executionStartToCloseTimeout,
-      String taskList,
-      Duration taskStartToCloseTimeout,
-      ChildPolicy childPolicy) {
+      Duration executionStartToCloseTimeout, String taskList, Duration taskStartToCloseTimeout) {
     this.executionStartToCloseTimeout = executionStartToCloseTimeout;
     this.taskList = taskList;
     this.taskStartToCloseTimeout = taskStartToCloseTimeout;
-    this.childPolicy = childPolicy;
   }
 
   public Duration getExecutionStartToCloseTimeout() {
@@ -86,8 +73,4 @@ public String getTaskList() {
   public Duration getTaskStartToCloseTimeout() {
     return taskStartToCloseTimeout;
   }
-
-  public ChildPolicy getChildPolicy() {
-    return childPolicy;
-  }
 }

src/main/java/com/uber/cadence/workflow/WorkflowInfo.java

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.workflow;
 
+import com.uber.cadence.ChildPolicy;
 import java.time.Duration;
 
 public interface WorkflowInfo {
@@ -32,4 +33,6 @@ public interface WorkflowInfo {
   String getTaskList();
 
   Duration getExecutionStartToCloseTimeout();
+
+  ChildPolicy getChildPolicy();
 }

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -26,10 +26,7 @@
 import static org.junit.Assert.fail;
 
 import com.google.common.util.concurrent.UncheckedExecutionException;
-import com.uber.cadence.SignalExternalWorkflowExecutionFailedCause;
-import com.uber.cadence.TimeoutType;
-import com.uber.cadence.WorkflowExecution;
-import com.uber.cadence.WorkflowIdReusePolicy;
+import com.uber.cadence.*;
 import com.uber.cadence.activity.Activity;
 import com.uber.cadence.activity.ActivityMethod;
 import com.uber.cadence.activity.ActivityOptions;
@@ -804,29 +801,42 @@ public void testDetachedScope() {
   public interface TestContinueAsNew {
 
     @WorkflowMethod
-    int execute(int count);
+    int execute(int count, String continueAsNewTaskList);
   }
 
   public static class TestContinueAsNewImpl implements TestContinueAsNew {
 
     @Override
-    public int execute(int count) {
+    public int execute(int count, String continueAsNewTaskList) {
+      String taskList = Workflow.getWorkflowInfo().getTaskList();
       if (count == 0) {
+        assertEquals(continueAsNewTaskList, taskList);
         return 111;
       }
-      TestContinueAsNew next = Workflow.newContinueAsNewStub(TestContinueAsNew.class, null);
-      next.execute(count - 1);
+      ContinueAsNewOptions options =
+          new ContinueAsNewOptions.Builder().setTaskList(continueAsNewTaskList).build();
+      TestContinueAsNew next = Workflow.newContinueAsNewStub(TestContinueAsNew.class, options);
+      next.execute(count - 1, continueAsNewTaskList);
       throw new RuntimeException("unreachable");
     }
   }
 
   @Test
   public void testContinueAsNew() {
+    Worker w2;
+    String continuedTaskList = this.taskList + "_continued";
+    if (useExternalService) {
+      w2 = workerFactory.newWorker(continuedTaskList);
+    } else {
+      w2 = testEnvironment.newWorker(continuedTaskList);
+    }
+    w2.registerWorkflowImplementationTypes(TestContinueAsNewImpl.class);
     startWorkerFor(TestContinueAsNewImpl.class);
+
     TestContinueAsNew client =
         workflowClient.newWorkflowStub(
-            TestContinueAsNew.class, newWorkflowOptionsBuilder(taskList).build());
-    int result = client.execute(4);
+            TestContinueAsNew.class, newWorkflowOptionsBuilder(this.taskList).build());
+    int result = client.execute(4, continuedTaskList);
     assertEquals(111, result);
     tracer.setExpected("continueAsNew", "continueAsNew", "continueAsNew", "continueAsNew");
   }
@@ -1984,6 +1994,7 @@ public void testSignalUntyped() {
         () -> {
           assertEquals("initial", client.query("QueryableWorkflow::getState", String.class));
           client.signal("testSignal", "Hello ");
+          sleep(Duration.ofMillis(500));
           while (!"Hello ".equals(client.query("QueryableWorkflow::getState", String.class))) {}
           assertEquals("Hello ", client.query("QueryableWorkflow::getState", String.class));
           client.signal("testSignal", "World!");