Cleanup threads created in signal method on decider close (#405)

Author: meiliang86

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

@@ -19,7 +19,17 @@
 
 import static com.uber.cadence.internal.common.InternalUtils.createStickyTaskList;
 
-import com.uber.cadence.*;
+import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
+import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
+import com.uber.cadence.HistoryEvent;
+import com.uber.cadence.PollForDecisionTaskResponse;
+import com.uber.cadence.QueryTaskCompletedType;
+import com.uber.cadence.RespondDecisionTaskCompletedRequest;
+import com.uber.cadence.RespondDecisionTaskFailedRequest;
+import com.uber.cadence.RespondQueryTaskCompletedRequest;
+import com.uber.cadence.StickyExecutionAttributes;
+import com.uber.cadence.WorkflowExecution;
+import com.uber.cadence.WorkflowType;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.worker.DecisionTaskHandler;

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

@@ -320,6 +320,11 @@ public void close() {
       return;
     }
     try {
+      for (WorkflowThread c : threadsToAdd) {
+        threads.addLast(c);
+      }
+      threadsToAdd.clear();
+
       for (WorkflowThread c : threads) {
         threadFutures.add(c.stopNow());
       }

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

@@ -5194,6 +5194,63 @@ public void testSagaParallelCompensation() {
     assertTrue(tracer.getTrace().contains("executeActivity TestActivities::activity2"));
   }
 
+  public static class TestSignalExceptionWorkflowImpl implements TestWorkflowSignaled {
+    private boolean signaled = false;
+
+    @Override
+    public String execute() {
+      Workflow.await(() -> signaled);
+      return null;
+    }
+
+    @Override
+    public void signal1(String arg) {
+      for (int i = 0; i < 100; i++) {
+        Async.procedure(() -> System.out.println("test"));
+      }
+
+      throw new RuntimeException("exception in signal method");
+    }
+  }
+
+  @Test
+  public void testExceptionInSignal() throws InterruptedException {
+    startWorkerFor(TestSignalExceptionWorkflowImpl.class);
+    TestWorkflowSignaled signalWorkflow =
+        workflowClient.newWorkflowStub(
+            TestWorkflowSignaled.class, newWorkflowOptionsBuilder(taskList).build());
+    CompletableFuture<String> result = WorkflowClient.execute(signalWorkflow::execute);
+    signalWorkflow.signal1("test");
+    try {
+      result.get(1, TimeUnit.SECONDS);
+      fail("not reachable");
+    } catch (Exception e) {
+      // exception expected here.
+    }
+
+    // Suspend polling so that decision tasks are not retried. Otherwise it will affect our thread
+    // count.
+    if (useExternalService) {
+      workerFactory.suspendPolling();
+    } else {
+      testEnvironment.getWorkerFactory().suspendPolling();
+    }
+
+    // Wait for decision task retry to finish.
+    Thread.sleep(10000);
+
+    int workflowThreads = 0;
+    ThreadInfo[] threads = ManagementFactory.getThreadMXBean().dumpAllThreads(false, false);
+    for (ThreadInfo thread : threads) {
+      if (thread.getThreadName().startsWith("workflow")) {
+        workflowThreads++;
+      }
+    }
+
+    assertTrue(
+        "workflow threads might leak, #workflowThreads = " + workflowThreads, workflowThreads < 20);
+  }
+
   private static class TracingWorkflowInterceptor implements WorkflowInterceptor {
 
     private final FilteredTrace trace;