Bug bash feedback (#111)

* Implemented options.getMaxWorkflowThreads value propagation to executor

* Pull request comments

Author: mfateev

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

@@ -25,6 +25,7 @@
 import com.uber.cadence.internal.worker.WorkflowWorker;
 
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
@@ -38,9 +39,10 @@ public class SyncWorkflowWorker {
     private final POJOWorkflowImplementationFactory factory;
     private final SingleWorkerOptions options;
 
-    public SyncWorkflowWorker(WorkflowService.Iface service, String domain, String taskList, SingleWorkerOptions options) {
-        ThreadPoolExecutor workflowThreadPool = new ThreadPoolExecutor(1000, 1000,
-                10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000));
+    public SyncWorkflowWorker(WorkflowService.Iface service, String domain, String taskList, SingleWorkerOptions options,
+                              int workflowThreadPoolSize) {
+        ThreadPoolExecutor workflowThreadPool = new ThreadPoolExecutor(workflowThreadPoolSize, workflowThreadPoolSize,
+                10, TimeUnit.SECONDS, new SynchronousQueue<>());
         factory = new POJOWorkflowImplementationFactory(options.getDataConverter(), workflowThreadPool);
         taskHandler = new ReplayDecisionTaskHandler(domain, factory);
         worker = new WorkflowWorker(service, domain, taskList, options, taskHandler);

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

@@ -27,6 +27,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -188,7 +189,13 @@ public void start() {
         }
         log.trace(String.format("Workflow thread \"%s\" started", getName()));
         context.setStatus(Status.RUNNING);
-        taskFuture = threadPool.submit(task);
+        try {
+            taskFuture = threadPool.submit(task);
+        } catch (RejectedExecutionException e) {
+            throw new Error("Not enough threads to execute workflows. " +
+                    "If this message appears consistently either WorkerOptions.maxConcurrentWorklfowExecutionSize " +
+                    "should be decreased or WorkerOptions.maxWorkflowThreads increased.");
+        }
     }
 
     public WorkflowThreadContext getContext() {

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

@@ -21,6 +21,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -118,9 +120,12 @@ public void start() {
                     options.getMaximumPollRateIntervalMilliseconds());
         }
 
+        // It is important to pass blocking queue of at least options.getPollThreadCount() capacity.
+        // As task enqueues next task the buffering is needed to queue task until the previous one
+        // releases a thread.
         pollExecutor = new ThreadPoolExecutor(options.getPollThreadCount(), options.getPollThreadCount(),
                 1, TimeUnit.SECONDS,
-                new LinkedBlockingQueue<>(options.getPollThreadCount()));
+                new ArrayBlockingQueue<>(options.getPollThreadCount()));
         pollExecutor.setThreadFactory(new ExecutorThreadFactory(options.getPollThreadNamePrefix(),
                 options.getUncaughtExceptionHandler()));
 

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

@@ -20,7 +20,7 @@
 import java.time.Duration;
 
 /**
- * TODO: Switch to Duration.
+ * Options for component that polls Cadence task lists for tasks.
  */
 public final class PollerOptions {
 
@@ -59,41 +59,66 @@ public Builder(PollerOptions o) {
             this.uncaughtExceptionHandler = o.getUncaughtExceptionHandler();
         }
 
+        /**
+         * Defines interval for measuring poll rate. Larger the interval more spiky can be the load.
+         */
         public Builder setMaximumPollRateIntervalMilliseconds(int maximumPollRateIntervalMilliseconds) {
             this.maximumPollRateIntervalMilliseconds = maximumPollRateIntervalMilliseconds;
             return this;
         }
 
+        /**
+         * Maximum rate of polling. Measured in the interval set through {@link #setMaximumPollRateIntervalMilliseconds(int)}.
+         */
         public Builder setMaximumPollRatePerSecond(double maximumPollRatePerSecond) {
             this.maximumPollRatePerSecond = maximumPollRatePerSecond;
             return this;
         }
 
+        /**
+         * Coefficient to use when calculating exponential delay in case of failures
+         */
         public Builder setPollBackoffCoefficient(double pollBackoffCoefficient) {
             this.pollBackoffCoefficient = pollBackoffCoefficient;
             return this;
         }
 
+        /**
+         * Initial delay in case of failure. If backoff coefficient is 1 then it would be the constant delay
+         * between failing polls.
+         */
         public Builder setPollBackoffInitialInterval(Duration pollBackoffInitialInterval) {
             this.pollBackoffInitialInterval = pollBackoffInitialInterval;
             return this;
         }
 
+        /**
+         * Maximum interval between polls in case of failures.
+         */
         public Builder setPollBackoffMaximumInterval(Duration pollBackoffMaximumInterval) {
             this.pollBackoffMaximumInterval = pollBackoffMaximumInterval;
             return this;
         }
 
+        /**
+         * Number of parallel polling threads.
+         */
         public Builder setPollThreadCount(int pollThreadCount) {
             this.pollThreadCount = pollThreadCount;
             return this;
         }
 
+        /**
+         * Called to report unexpected exceptions in the poller threads.
+         */
         public Builder setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler uncaughtExceptionHandler) {
             this.uncaughtExceptionHandler = uncaughtExceptionHandler;
             return this;
         }
 
+        /**
+         * Prefix to use when naming poller threads.
+         */
         public Builder setPollThreadNamePrefix(String pollThreadNamePrefix) {
             this.pollThreadNamePrefix = pollThreadNamePrefix;
             return this;

src/main/java/com/uber/cadence/worker/Worker.java

@@ -105,7 +105,8 @@ public Worker(WorkflowService.Iface service, String domain, String taskList, Wor
         }
         SingleWorkerOptions workflowOptions = toWorkflowOptions(options);
         if (!options.isDisableWorkflowWorker()) {
-            workflowWorker = new SyncWorkflowWorker(service, domain, taskList, workflowOptions);
+            workflowWorker = new SyncWorkflowWorker(service, domain, taskList, workflowOptions,
+                    options.getMaxWorkflowThreads());
         } else {
             workflowWorker = null;
         }
@@ -134,7 +135,7 @@ private SingleWorkerOptions toWorkflowOptions(WorkerOptions options) {
     }
 
     /**
-     * Register workflow implementation classes with a worker.
+     * Register workflow implementation classes with a worker. Overwrites previously registered types.
      * A workflow implementation class must implement at least one interface with a method annotated with
      * {@link com.uber.cadence.workflow.WorkflowMethod}. That method becomes a workflow type
      * that this worker supports.
@@ -151,7 +152,7 @@ public void registerWorkflowImplementationTypes(Class<?>... workflowImplementati
     }
 
     /**
-     * Register activity implementation objects with a worker.
+     * Register activity implementation objects with a worker. Overwrites previously registered objects.
      * As activities are reentrant and stateless only one instance per activity type
      * is registered.
      * <p>
@@ -164,11 +165,11 @@ public void registerActivitiesImplementations(Object... activityImplementations)
         if (activityWorker == null) {
             throw new IllegalStateException("disableActivityWorker is set in worker options");
         }
-        checkStarted();
+        checkNotStarted();
         activityWorker.setActivitiesImplementation(activityImplementations);
     }
 
-    private void checkStarted() {
+    private void checkNotStarted() {
         if (started.get()) {
             throw new IllegalStateException("already started");
         }