Added simulated timeout support for a child workflow (#139)

* Fxed support for mocked activity implementation registration

* Added a check when a child workflow stub is used to signal only

* Added simulated timout to child workflow

Author: mfateev

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

@@ -35,7 +35,7 @@ class ChildWorkflowStubImpl implements ChildWorkflowStub {
   private final String workflowType;
   private final ChildWorkflowOptions options;
   private final WorkflowInterceptor decisionContext;
-  private CompletablePromise<WorkflowExecution> execution = Workflow.newPromise();
+  private CompletablePromise<WorkflowExecution> execution;
 
   ChildWorkflowStubImpl(
       String workflowType, ChildWorkflowOptions options, WorkflowInterceptor decisionContext) {
@@ -80,20 +80,28 @@ public <R> R execute(Class<R> returnType, Object... args) {
   public <R> Promise<R> executeAsync(Class<R> returnType, Object... args) {
     WorkflowResult<R> result =
         decisionContext.executeChildWorkflow(workflowType, returnType, args, options);
+    execution = Workflow.newPromise();
     execution.completeFrom(result.getWorkflowExecution());
     return result.getResult();
   }
 
   @Override
   public void signal(String signalName, Object... args) {
-    Promise<Void> signalled =
+    if (execution == null) {
+      throw new IllegalStateException(
+          "This stub cannot be used to signal a workflow"
+              + " without starting it first. "
+              + "To signal a workflow execution that was started elsewhere "
+              + "use a stub created through Workflow.newExternalWorkflowStub");
+    }
+    Promise<Void> signaled =
         decisionContext.signalExternalWorkflow(execution.get(), signalName, args);
     if (AsyncInternal.isAsync()) {
-      AsyncInternal.setAsyncResult(signalled);
+      AsyncInternal.setAsyncResult(signaled);
       return;
     }
     try {
-      signalled.get();
+      signaled.get();
     } catch (SignalExternalWorkflowException e) {
       // Reset stack to the current one. Otherwise it is very confusing to see a stack of
       // an event handling method.

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

@@ -44,13 +44,13 @@ public WorkflowExecution getExecution() {
 
   @Override
   public void signal(String signalName, Object... args) {
-    Promise<Void> signalled = decisionContext.signalExternalWorkflow(execution, signalName, args);
+    Promise<Void> signaled = decisionContext.signalExternalWorkflow(execution, signalName, args);
     if (AsyncInternal.isAsync()) {
-      AsyncInternal.setAsyncResult(signalled);
+      AsyncInternal.setAsyncResult(signaled);
       return;
     }
     try {
-      signalled.get();
+      signaled.get();
     } catch (SignalExternalWorkflowException e) {
       // Reset stack to the current one. Otherwise it is very confusing to see a stack of
       // an event handling method.

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

@@ -31,7 +31,7 @@
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.worker.ActivityTaskHandler;
 import com.uber.cadence.serviceclient.IWorkflowService;
-import com.uber.cadence.testing.TestActivityTimeoutException;
+import com.uber.cadence.testing.SimulatedTimeoutException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
@@ -78,6 +78,9 @@ public void addActivityImplementation(Object activity) {
       throw new IllegalArgumentException("Activity must implement at least one interface");
     }
     for (TypeToken<?> i : interfaces) {
+      if (i.getType().getTypeName().startsWith("org.mockito")) {
+        continue;
+      }
       for (Method method : i.getRawType().getMethods()) {
         POJOActivityImplementation implementation =
             new POJOActivityImplementation(method, activity);
@@ -105,10 +108,10 @@ private ActivityTaskHandler.Result mapToActivityFailure(ActivityTask task, Throw
       throw new CancellationException(failure.getMessage());
     }
     // Only expected during unit tests.
-    if (failure instanceof TestActivityTimeoutException) {
-      TestActivityTimeoutException timeoutException = (TestActivityTimeoutException) failure;
+    if (failure instanceof SimulatedTimeoutException) {
+      SimulatedTimeoutException timeoutException = (SimulatedTimeoutException) failure;
       failure =
-          new TestActivityTimeoutExceptionInternal(
+          new SimulatedTimeoutExceptionInternal(
               timeoutException.getTimeoutType(),
               dataConverter.toData(timeoutException.getDetails()));
     }

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

@@ -26,6 +26,7 @@
 import com.uber.cadence.internal.replay.ReplayWorkflow;
 import com.uber.cadence.internal.replay.ReplayWorkflowFactory;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
+import com.uber.cadence.testing.SimulatedTimeoutException;
 import com.uber.cadence.workflow.Functions;
 import com.uber.cadence.workflow.QueryMethod;
 import com.uber.cadence.workflow.SignalMethod;
@@ -291,6 +292,15 @@ public void processSignal(String signalName, byte[] input, long eventId) {
   public static WorkflowExecutionException mapToWorkflowExecutionException(
       Exception failure, DataConverter dataConverter) {
     failure = CheckedExceptionWrapper.unwrap(failure);
+    // Only expected during unit tests.
+    if (failure instanceof SimulatedTimeoutException) {
+      SimulatedTimeoutException timeoutException = (SimulatedTimeoutException) failure;
+      failure =
+          new SimulatedTimeoutExceptionInternal(
+              timeoutException.getTimeoutType(),
+              dataConverter.toData(timeoutException.getDetails()));
+    }
+
     return new WorkflowExecutionException(
         failure.getClass().getName(), dataConverter.toData(failure));
   }

src/main/java/com/uber/cadence/internal/sync/TestActivityTimeoutExceptionInternal.java renamed to src/main/java/com/uber/cadence/internal/sync/SimulatedTimeoutExceptionInternal.java

@@ -20,21 +20,21 @@
 import com.uber.cadence.TimeoutType;
 
 /**
- * TestActivityTimeoutException is created from an TestActivityTimeoutException. The main difference
- * is that details are in a serialized form.
+ * SimulatedTimeoutExceptionInternal is created from a SimulatedTimeoutException. The main
+ * difference is that the details are in a serialized form.
  */
-final class TestActivityTimeoutExceptionInternal extends RuntimeException {
+final class SimulatedTimeoutExceptionInternal extends RuntimeException {
 
   private final TimeoutType timeoutType;
 
   private final byte[] details;
 
-  TestActivityTimeoutExceptionInternal(TimeoutType timeoutType, byte[] details) {
+  SimulatedTimeoutExceptionInternal(TimeoutType timeoutType, byte[] details) {
     this.timeoutType = timeoutType;
     this.details = details;
   }
 
-  TestActivityTimeoutExceptionInternal(TimeoutType timeoutType) {
+  SimulatedTimeoutExceptionInternal(TimeoutType timeoutType) {
     this.timeoutType = timeoutType;
     this.details = null;
   }

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

@@ -39,6 +39,7 @@
 import com.uber.cadence.workflow.ChildWorkflowException;
 import com.uber.cadence.workflow.ChildWorkflowFailureException;
 import com.uber.cadence.workflow.ChildWorkflowOptions;
+import com.uber.cadence.workflow.ChildWorkflowTimedOutException;
 import com.uber.cadence.workflow.CompletablePromise;
 import com.uber.cadence.workflow.ContinueAsNewOptions;
 import com.uber.cadence.workflow.Functions;
@@ -180,10 +181,9 @@ private RuntimeException mapActivityException(Exception failure) {
       } catch (Exception e) {
         cause = e;
       }
-      if (cause instanceof TestActivityTimeoutExceptionInternal) {
+      if (cause instanceof SimulatedTimeoutExceptionInternal) {
         // This exception is thrown only in unit tests to mock the activity timeouts
-        TestActivityTimeoutExceptionInternal testTimeout =
-            (TestActivityTimeoutExceptionInternal) cause;
+        SimulatedTimeoutExceptionInternal testTimeout = (SimulatedTimeoutExceptionInternal) cause;
         return new ActivityTimeoutException(
             taskFailed.getEventId(),
             taskFailed.getActivityType(),
@@ -303,6 +303,11 @@ private RuntimeException mapChildWorkflowException(Exception failure) {
     } catch (Exception e) {
       cause = e;
     }
+    if (cause instanceof SimulatedTimeoutExceptionInternal) {
+      // This exception is thrown only in unit tests to mock the child workflow timeouts
+      return new ChildWorkflowTimedOutException(
+          taskFailed.getEventId(), taskFailed.getWorkflowExecution(), taskFailed.getWorkflowType());
+    }
     return new ChildWorkflowFailureException(
         taskFailed.getEventId(),
         taskFailed.getWorkflowExecution(),

src/main/java/com/uber/cadence/testing/TestActivityTimeoutException.java renamed to src/main/java/com/uber/cadence/testing/SimulatedTimeoutException.java

@@ -20,22 +20,29 @@
 import com.uber.cadence.TimeoutType;
 
 /**
- * TestActivityTimeoutException can be thrown from an activity implementation to simulate an
- * activity timeout. To be used only in unit tests. The workflow code is going to receive it as
- * {@link com.uber.cadence.workflow.ActivityTimeoutException}.
+ * SimulatedTimeoutException can be thrown from an activity or child workflow implementation to
+ * simulate a timeout. To be used only in unit tests. If thrown from an activity the workflow code
+ * is going to receive it as {@link com.uber.cadence.workflow.ActivityTimeoutException}. If thrown
+ * from a child workflow the workflow code is going to receive it as {@link
+ * com.uber.cadence.workflow.ChildWorkflowTimedOutException}.
  */
-public final class TestActivityTimeoutException extends RuntimeException {
+public final class SimulatedTimeoutException extends RuntimeException {
 
   private final TimeoutType timeoutType;
 
   private final Object details;
 
-  public TestActivityTimeoutException(TimeoutType timeoutType, Object details) {
+  public SimulatedTimeoutException(TimeoutType timeoutType, Object details) {
     this.timeoutType = timeoutType;
     this.details = details;
   }
 
-  public TestActivityTimeoutException(TimeoutType timeoutType) {
+  public SimulatedTimeoutException() {
+    this.timeoutType = TimeoutType.START_TO_CLOSE;
+    this.details = null;
+  }
+
+  public SimulatedTimeoutException(TimeoutType timeoutType) {
     this.timeoutType = timeoutType;
     this.details = null;
   }

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

@@ -58,8 +58,9 @@ public static ActivityStub newUntypedActivityStub(ActivityOptions options) {
   }
 
   /**
-   * Creates client stub that can be used to start a child workflow that implements given interface
-   * using parent options.
+   * Creates client stub that can be used to start a child workflow that implements the given
+   * interface using parent options. Use {@link #newExternalWorkflowStub(Class, String)} to get a
+   * stub to signal a workflow without starting it.
    *
    * @param workflowInterface interface type implemented by activities
    */
@@ -69,6 +70,8 @@ public static <T> T newChildWorkflowStub(Class<T> workflowInterface) {
 
   /**
    * Creates client stub that can be used to start a child workflow that implements given interface.
+   * Use {@link #newExternalWorkflowStub(Class, String)} to get a stub to signal a workflow without
+   * starting it.
    *
    * @param workflowInterface interface type implemented by activities
    * @param options options passed to the child workflow.

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

@@ -23,7 +23,7 @@
 
 /**
  * A value that is local to a single workflow execution. So it can act as a <i>global</i> variable
- * for the workflow code. For example the {@code Workflow.isSignalled()} static method returns the
+ * for the workflow code. For example the {@code Workflow.isSignaled()} static method returns the
  * correct value even if there are multiple workflows executing on the same machine simultaneously.
  * It would be invalid if the {@code signaled} was a {@code static boolean} variable.
  *
@@ -32,7 +32,7 @@
  *
  *   private static final WorkflowLocal<Boolean> signaled = WorkflowLocal.withInitial(() -> false);
  *
- *   public static boolean isSignalled() {
+ *   public static boolean isSignaled() {
  *     return signaled.get();
  *   }
  *

src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java

@@ -36,11 +36,12 @@
 import com.uber.cadence.client.WorkflowStub;
 import com.uber.cadence.client.WorkflowTimedOutException;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
-import com.uber.cadence.testing.TestActivityTimeoutException;
+import com.uber.cadence.testing.SimulatedTimeoutException;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
 import com.uber.cadence.worker.Worker;
 import com.uber.cadence.workflow.ActivityTimeoutException;
 import com.uber.cadence.workflow.Async;
+import com.uber.cadence.workflow.ChildWorkflowTimedOutException;
 import com.uber.cadence.workflow.Promise;
 import com.uber.cadence.workflow.SignalMethod;
 import com.uber.cadence.workflow.Workflow;
@@ -205,7 +206,7 @@ private static class SimulatedTimeoutActivityImpl implements TestActivity {
 
     @Override
     public String activity1(String input) {
-      throw new TestActivityTimeoutException(TimeoutType.HEARTBEAT, "progress1");
+      throw new SimulatedTimeoutException(TimeoutType.HEARTBEAT, "progress1");
     }
   }
 
@@ -619,4 +620,44 @@ public void testChild() {
     String result = workflow.workflow("input1");
     assertEquals("child input1", result);
   }
+
+  public static class SimulatedTimeoutParentWorkflow implements ParentWorkflow {
+
+    @Override
+    public String workflow(String input) {
+      ChildWorkflow child = Workflow.newChildWorkflowStub(ChildWorkflow.class);
+      Promise<String> result =
+          Async.function(child::workflow, input, Workflow.getWorkflowInfo().getWorkflowId());
+      return result.get();
+    }
+
+    @Override
+    public void signal(String value) {}
+  }
+
+  public static class SimulatedTimeoutChildWorklfow implements ChildWorkflow {
+
+    @Override
+    public String workflow(String input, String parentId) {
+      Workflow.sleep(Duration.ofHours(2));
+      throw new SimulatedTimeoutException();
+    }
+  }
+
+  @Test
+  public void testChildSimulatedTimeout() {
+    Worker worker = testEnvironment.newWorker(TASK_LIST);
+    worker.registerWorkflowImplementationTypes(
+        SimulatedTimeoutParentWorkflow.class, SimulatedTimeoutChildWorklfow.class);
+    worker.start();
+    WorkflowClient client = testEnvironment.newWorkflowClient();
+    WorkflowOptions options = new WorkflowOptions.Builder().setWorkflowId("parent1").build();
+    ParentWorkflow workflow = client.newWorkflowStub(ParentWorkflow.class, options);
+    try {
+      workflow.workflow("input1");
+      fail("unreacheable");
+    } catch (WorkflowException e) {
+      assertTrue(e.getCause() instanceof ChildWorkflowTimedOutException);
+    }
+  }
 }