Fix thread leak on non-deterministic error (#365)

* Also throw error on rejected execution exception

Author: meiliang86

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

@@ -168,6 +168,13 @@ private Result processDecision(PollForDecisionTaskResponse decisionTask) throws
       }
       return createCompletedRequest(decisionTask, result);
     } catch (Throwable e) {
+      // Note here that the decider might not be in the cache, even sticky is on. In that case we
+      // need to close the decider explicitly.
+      // For items in the cache, invalidation callback will try to close again, which should be ok.
+      if (decider != null) {
+        decider.close();
+      }
+
       if (stickyTaskListName != null) {
         cache.invalidate(decisionTask.getWorkflowExecution().getRunId());
       }

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

@@ -0,0 +1,25 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.sync;
+
+public class WorkflowRejectedExecutionError extends Error {
+
+  WorkflowRejectedExecutionError(Throwable cause) {
+    super(cause);
+  }
+}

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

@@ -256,10 +256,13 @@ public void start() {
               cache.evictAnyNotInProcessing(
                   this.runner.getDecisionContext().getContext().getRunId());
           if (!evicted) {
-            throw e;
+            // Note here we need to throw error, not exception. Otherwise it will be
+            // translated to workflow execution exception and instead of failing the
+            // decision we will be failing the workflow.
+            throw new WorkflowRejectedExecutionError(e);
           }
         } else {
-          throw e;
+          throw new WorkflowRejectedExecutionError(e);
         }
       }
     }

src/test/java/com/uber/cadence/internal/sync/DeterministicRunnerTest.java

@@ -809,4 +809,28 @@ public void close() {
       runner.close();
     }
   }
+
+  @Test
+  public void testRejectedExecutionError() {
+    ThreadPoolExecutor threadPool =
+        new ThreadPoolExecutor(0, 1, 1, TimeUnit.SECONDS, new SynchronousQueue<>());
+
+    DeterministicRunner d =
+        new DeterministicRunnerImpl(
+            threadPool,
+            null,
+            System::currentTimeMillis,
+            () -> {
+              Promise<Void> async = Async.procedure(() -> status = "started");
+              async.get();
+            });
+
+    assertEquals("initial", status);
+
+    try {
+      d.runUntilAllBlocked();
+    } catch (Throwable t) {
+      assertTrue(t instanceof WorkflowRejectedExecutionError);
+    }
+  }
 }

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

@@ -75,6 +75,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
 import java.lang.reflect.Type;
 import java.text.SimpleDateFormat;
 import java.time.Duration;
@@ -4176,7 +4177,7 @@ public void testNonDeterministicWorkflowPolicyBlockWorkflow() {
     startWorkerFor(DeterminismFailingWorkflowImpl.class);
     WorkflowOptions options =
         new WorkflowOptions.Builder()
-            .setExecutionStartToCloseTimeout(Duration.ofSeconds(1))
+            .setExecutionStartToCloseTimeout(Duration.ofSeconds(10))
             .setTaskStartToCloseTimeout(Duration.ofSeconds(1))
             .setTaskList(taskList)
             .build();
@@ -4188,6 +4189,16 @@ public void testNonDeterministicWorkflowPolicyBlockWorkflow() {
     } catch (WorkflowTimedOutException e) {
       // expected to timeout as workflow is going get blocked.
     }
+
+    int workflowRootThreads = 0;
+    ThreadInfo[] threads = ManagementFactory.getThreadMXBean().dumpAllThreads(false, false);
+    for (ThreadInfo thread : threads) {
+      if (thread.getThreadName().contains("workflow-root")) {
+        workflowRootThreads++;
+      }
+    }
+
+    assertTrue("workflow threads might leak", workflowRootThreads < 10);
   }
 
   @Test