Thread leaks2 (#186)

* Fixed core sizes of thread pools
* Increased WorkflowTest[TestService] timeout to not have failures on travis

* Increased WorkflowTest[TestService] timeout to not have failures on travis

* Added thread dump after the fixed timeout

* Updated thread dump delay

* Fixed race condition in TestWorkflowMutableState.query

* Disabled excessive unit test logging

Author: mfateev

build.gradle

@@ -215,3 +215,5 @@ uploadArchives {
         }
     }
 }
+
+//logging.captureStandardOutput LogLevel.INFO

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

@@ -45,10 +45,7 @@
 import java.util.Random;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiConsumer;
@@ -121,11 +118,20 @@ static void setCurrentThreadInternal(WorkflowThread coroutine) {
   }
 
   DeterministicRunnerImpl(Supplier<Long> clock, Runnable root) {
-    this(
-        new ThreadPoolExecutor(0, 1000, 1, TimeUnit.MINUTES, new SynchronousQueue<>()),
-        newDummySyncDecisionContext(),
-        clock,
-        root);
+    this(getDefaultThreadPool(), newDummySyncDecisionContext(), clock, root);
+  }
+
+  private static ThreadPoolExecutor getDefaultThreadPool() {
+    ThreadPoolExecutor result =
+        new ThreadPoolExecutor(0, 1000, 1, TimeUnit.SECONDS, new SynchronousQueue<>());
+    result.setThreadFactory(
+        new ThreadFactory() {
+          @Override
+          public Thread newThread(Runnable r) {
+            return new Thread(r, "deterministic runner thread");
+          }
+        });
+    return result;
   }
 
   DeterministicRunnerImpl(

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

@@ -30,6 +30,7 @@
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 
 /** Workflow worker that supports POJO workflow implementations. */
@@ -38,6 +39,7 @@ public class SyncWorkflowWorker {
   private final WorkflowWorker worker;
   private final POJOWorkflowImplementationFactory factory;
   private final SingleWorkerOptions options;
+  private final AtomicInteger workflowThreadCounter = new AtomicInteger();
 
   public SyncWorkflowWorker(
       IWorkflowService service,
@@ -48,11 +50,9 @@ public SyncWorkflowWorker(
       int workflowThreadPoolSize) {
     ThreadPoolExecutor workflowThreadPool =
         new ThreadPoolExecutor(
-            workflowThreadPoolSize,
-            workflowThreadPoolSize,
-            10,
-            TimeUnit.SECONDS,
-            new SynchronousQueue<>());
+            0, workflowThreadPoolSize, 1, TimeUnit.SECONDS, new SynchronousQueue<>());
+    workflowThreadPool.setThreadFactory(
+        r -> new Thread(r, "workflow-thread-" + workflowThreadCounter.incrementAndGet()));
     factory =
         new POJOWorkflowImplementationFactory(
             options.getDataConverter(),

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -1173,9 +1173,9 @@ public QueryWorkflowResponse query(QueryWorkflowRequest queryRequest) throws TEx
             .setQuery(queryRequest.getQuery());
     TaskListId taskListId =
         new TaskListId(queryRequest.getDomain(), startRequest.getTaskList().getName());
-    store.sendQueryTask(executionId, taskListId, task);
     CompletableFuture<QueryWorkflowResponse> result = new CompletableFuture<>();
     queries.put(queryId.getQueryId(), result);
+    store.sendQueryTask(executionId, taskListId, task);
     try {
       return result.get();
     } catch (InterruptedException e) {

src/main/java/com/uber/cadence/internal/worker/PollTask.java

@@ -65,9 +65,9 @@ public interface Poller<T> {
     this.handler = handler;
     taskExecutor =
         new ThreadPoolExecutor(
+            0,
             options.getTaskExecutorThreadPoolSize(),
-            options.getTaskExecutorThreadPoolSize(),
-            10,
+            1,
             TimeUnit.SECONDS,
             new SynchronousQueue<>());
     taskExecutor.setThreadFactory(

src/main/java/com/uber/cadence/internal/worker/Poller.java

@@ -180,22 +180,17 @@ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedE
       // not started yet.
       return true;
     }
-    boolean result = pollExecutor.awaitTermination(timeout, unit);
-    log.info("awaitTermination done");
-    return result;
+    return pollExecutor.awaitTermination(timeout, unit);
   }
 
   @Override
   public boolean shutdownAndAwaitTermination(long timeout, TimeUnit unit)
       throws InterruptedException {
-    log.info("shutdownAndAwaitTermination poller=" + this.options.getPollThreadNamePrefix());
     if (!isStarted()) {
       return true;
     }
     pollExecutor.shutdownNow();
-    boolean result = pollExecutor.awaitTermination(timeout, unit);
-    log.info("shutdownAndAwaitTermination done");
-    return result;
+    return pollExecutor.awaitTermination(timeout, unit);
   }
 
   @Override

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

@@ -25,7 +25,6 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import com.googlecode.junittoolbox.ParallelParameterized;
 import com.uber.cadence.SignalExternalWorkflowExecutionFailedCause;
 import com.uber.cadence.TimeoutType;
 import com.uber.cadence.WorkflowExecution;
@@ -60,20 +59,7 @@
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.time.Duration;
-import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Queue;
-import java.util.Random;
-import java.util.Set;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -99,12 +85,13 @@
 import org.junit.rules.Timeout;
 import org.junit.runner.Description;
 import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameter;
 import org.junit.runners.Parameterized.Parameters;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@RunWith(ParallelParameterized.class)
+@RunWith(Parameterized.class)
 public class WorkflowTest {
 
   /**
@@ -132,7 +119,7 @@ public static Object[] data() {
 
   @Rule
   public Timeout globalTimeout =
-      Timeout.seconds(DEBUGGER_TIMEOUTS ? 500 : (skipDockerService ? 3 : 20));
+      Timeout.seconds(DEBUGGER_TIMEOUTS ? 500 : (skipDockerService ? 5 : 20));
 
   @Rule
   public TestWatcher watchman =
@@ -251,7 +238,7 @@ public void setUp() {
 
   @After
   public void tearDown() throws Throwable {
-    worker.shutdown(Duration.ofMillis(1));
+    worker.shutdown(Duration.ofMinutes(1));
     activitiesImpl.close();
     if (testEnvironment != null) {
       testEnvironment.close();
@@ -2581,14 +2568,15 @@ private static class TestActivitiesImpl implements TestActivities {
     final List<String> invocations = Collections.synchronizedList(new ArrayList<>());
     final List<String> procResult = Collections.synchronizedList(new ArrayList<>());
     private final ThreadPoolExecutor executor =
-        new ThreadPoolExecutor(2, 100, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>());
+        new ThreadPoolExecutor(0, 100, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
 
     private TestActivitiesImpl(ActivityCompletionClient completionClient) {
       this.completionClient = completionClient;
     }
 
-    void close() {
-      executor.shutdown();
+    void close() throws InterruptedException {
+      executor.shutdownNow();
+      executor.awaitTermination(1, TimeUnit.MINUTES);
     }
 
     void assertInvocations(String... expected) {