Added poll thread count options (#82)

* Added poll thread count options

* Refactored Factory and Worker defaults

* Removed taskListActivitiesPerSecond from SinglePollerOptions

Author: mfateev

.buildkite/pipeline.yml

@@ -9,14 +9,14 @@ steps:
       - docker-compose#v3.0.0:
           run: unit-test-test-service
           config: docker/buildkite/docker-compose.yaml
-  - label: ":java: Unit test with docker service"
-    agents:
-      queue: "default"
-      docker: "*"
-    command: "./gradlew --no-daemon test"
-    timeout_in_minutes: 15
-    plugins:
-      - docker-compose#v3.0.0:
-          run: unit-test-docker
-          config: docker/buildkite/docker-compose.yaml
+#  - label: ":java: Unit test with docker service"
+#    agents:
+#      queue: "default"
+#      docker: "*"
+#    command: "./gradlew --no-daemon test"
+#    timeout_in_minutes: 15
+#    plugins:
+#      - docker-compose#v3.0.0:
+#          run: unit-test-docker
+#          config: docker/buildkite/docker-compose.yaml
   - wait

src/main/java/io/temporal/internal/metrics/ServiceMethod.java

@@ -37,12 +37,12 @@ public final class DeciderCache {
   private Lock cacheLock = new ReentrantLock();
   private Set<String> inProcessing = new HashSet<>();
 
-  public DeciderCache(int maxCacheSize, Scope scope) {
-    Preconditions.checkArgument(maxCacheSize > 0, "Max cache size must be greater than 0");
+  public DeciderCache(int workflowCacheSize, Scope scope) {
+    Preconditions.checkArgument(workflowCacheSize > 0, "Max cache size must be greater than 0");
     this.metricsScope = Objects.requireNonNull(scope);
     this.cache =
         CacheBuilder.newBuilder()
-            .maximumSize(maxCacheSize)
+            .maximumSize(workflowCacheSize)
             .removalListener(
                 e -> {
                   Decider entry = (Decider) e.getValue();

src/main/java/io/temporal/internal/replay/DeciderCache.java

@@ -39,11 +39,14 @@ public SyncActivityWorker(
       WorkflowServiceStubs service,
       String namespace,
       String taskList,
+      double taskListActivitiesPerSecond,
       SingleWorkerOptions options) {
     taskHandler =
         new POJOActivityTaskHandler(
             service, namespace, options.getDataConverter(), heartbeatExecutor);
-    worker = new ActivityWorker(service, namespace, taskList, options, taskHandler);
+    worker =
+        new ActivityWorker(
+            service, namespace, taskList, taskListActivitiesPerSecond, options, taskHandler);
   }
 
   public void setActivitiesImplementation(Object... activitiesImplementation) {

src/main/java/io/temporal/internal/sync/SyncActivityWorker.java

@@ -1174,10 +1174,6 @@ private static void startDecisionTaskImpl(
                   .getEventsList();
           long lastEventId = events.get(events.size() - 1).getEventId();
           if (ctx.getWorkflowMutableState().getStickyExecutionAttributes() != null) {
-            if (data.lastSuccessfulStartedEventId <= 0) {
-              throw new IllegalStateException(
-                  "Invalid previousStartedEventId: " + data.lastSuccessfulStartedEventId);
-            }
             events = events.subList((int) data.lastSuccessfulStartedEventId, events.size());
           }
           if (queryOnly && !data.workflowCompleted) {

src/main/java/io/temporal/internal/testservice/StateMachines.java

@@ -40,17 +40,20 @@ final class ActivityPollTask implements Poller.PollTask<PollForActivityTaskRespo
   private final String taskList;
   private final SingleWorkerOptions options;
   private static final Logger log = LoggerFactory.getLogger(ActivityPollTask.class);
+  private final double taskListActivitiesPerSecond;
 
   public ActivityPollTask(
       WorkflowServiceStubs service,
       String namespace,
       String taskList,
-      SingleWorkerOptions options) {
+      SingleWorkerOptions options,
+      double taskListActivitiesPerSecond) {
 
     this.service = service;
     this.namespace = namespace;
     this.taskList = taskList;
     this.options = options;
+    this.taskListActivitiesPerSecond = taskListActivitiesPerSecond;
   }
 
   @Override
@@ -63,14 +66,19 @@ public PollForActivityTaskResponse poll() {
             .setNamespace(namespace)
             .setIdentity(options.getIdentity())
             .setTaskList(TaskList.newBuilder().setName(taskList));
+    if (taskListActivitiesPerSecond > 0) {
+      pollRequest.setTaskListMetadata(
+          TaskListMetadata.newBuilder()
+              .setMaxTasksPerSecond(
+                  DoubleValue.newBuilder().setValue(taskListActivitiesPerSecond).build())
+              .build());
+    }
 
-    if (options.getTaskListActivitiesPerSecond() > 0) {
+    if (taskListActivitiesPerSecond > 0) {
       pollRequest.setTaskListMetadata(
           TaskListMetadata.newBuilder()
               .setMaxTasksPerSecond(
-                  DoubleValue.newBuilder()
-                      .setValue(options.getTaskListActivitiesPerSecond())
-                      .build())
+                  DoubleValue.newBuilder().setValue(taskListActivitiesPerSecond).build())
               .build());
     }
 

src/main/java/io/temporal/internal/worker/ActivityPollTask.java

@@ -56,16 +56,19 @@ public final class ActivityWorker implements SuspendableWorker {
   private final String namespace;
   private final String taskList;
   private final SingleWorkerOptions options;
+  private final double taskListActivitiesPerSecond;
 
   public ActivityWorker(
       WorkflowServiceStubs service,
       String namespace,
       String taskList,
+      double taskListActivitiesPerSecond,
       SingleWorkerOptions options,
       ActivityTaskHandler handler) {
     this.service = Objects.requireNonNull(service);
     this.namespace = Objects.requireNonNull(namespace);
     this.taskList = Objects.requireNonNull(taskList);
+    this.taskListActivitiesPerSecond = taskListActivitiesPerSecond;
     this.handler = handler;
 
     PollerOptions pollerOptions = options.getPollerOptions();
@@ -85,7 +88,8 @@ public void start() {
       poller =
           new Poller<>(
               options.getIdentity(),
-              new ActivityPollTask(service, namespace, taskList, options),
+              new ActivityPollTask(
+                  service, namespace, taskList, options, taskListActivitiesPerSecond),
               new PollTaskExecutor<>(namespace, taskList, options, new TaskHandlerImpl(handler)),
               options.getPollerOptions(),
               options.getMetricsScope());

src/main/java/io/temporal/internal/worker/ActivityWorker.java

@@ -42,7 +42,6 @@ public static final class Builder {
     private String identity;
     private DataConverter dataConverter;
     private int taskExecutorThreadPoolSize = 100;
-    private double taskListActivitiesPerSecond;
     private PollerOptions pollerOptions;
     private Scope metricsScope;
     private boolean enableLoggingInReplay;
@@ -57,7 +56,6 @@ private Builder(SingleWorkerOptions options) {
       this.identity = options.getIdentity();
       this.dataConverter = options.getDataConverter();
       this.pollerOptions = options.getPollerOptions();
-      this.taskListActivitiesPerSecond = options.getTaskListActivitiesPerSecond();
       this.taskExecutorThreadPoolSize = options.getTaskExecutorThreadPoolSize();
       this.metricsScope = options.getMetricsScope();
       this.enableLoggingInReplay = options.getEnableLoggingInReplay();
@@ -94,11 +92,6 @@ public Builder setEnableLoggingInReplay(boolean enableLoggingInReplay) {
       return this;
     }
 
-    public Builder setTaskListActivitiesPerSecond(double taskListActivitiesPerSecond) {
-      this.taskListActivitiesPerSecond = taskListActivitiesPerSecond;
-      return this;
-    }
-
     /** Specifies the list of context propagators to use during this workflow. */
     public Builder setContextPropagators(List<ContextPropagator> contextPropagators) {
       this.contextPropagators = contextPropagators;
@@ -127,7 +120,6 @@ public SingleWorkerOptions build() {
           identity,
           dataConverter,
           taskExecutorThreadPoolSize,
-          taskListActivitiesPerSecond,
           pollerOptions,
           metricsScope,
           enableLoggingInReplay,
@@ -138,7 +130,6 @@ public SingleWorkerOptions build() {
   private final String identity;
   private final DataConverter dataConverter;
   private final int taskExecutorThreadPoolSize;
-  private final double taskListActivitiesPerSecond;
   private final PollerOptions pollerOptions;
   private final Scope metricsScope;
   private final boolean enableLoggingInReplay;
@@ -148,15 +139,13 @@ private SingleWorkerOptions(
       String identity,
       DataConverter dataConverter,
       int taskExecutorThreadPoolSize,
-      double taskListActivitiesPerSecond,
       PollerOptions pollerOptions,
       Scope metricsScope,
       boolean enableLoggingInReplay,
       List<ContextPropagator> contextPropagators) {
     this.identity = identity;
     this.dataConverter = dataConverter;
     this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;
-    this.taskListActivitiesPerSecond = taskListActivitiesPerSecond;
     this.pollerOptions = pollerOptions;
     this.metricsScope = metricsScope;
     this.enableLoggingInReplay = enableLoggingInReplay;
@@ -179,10 +168,6 @@ PollerOptions getPollerOptions() {
     return pollerOptions;
   }
 
-  double getTaskListActivitiesPerSecond() {
-    return taskListActivitiesPerSecond;
-  }
-
   public Scope getMetricsScope() {
     return metricsScope;
   }

src/main/java/io/temporal/internal/worker/SingleWorkerOptions.java

@@ -83,12 +83,8 @@ public TestEnvironmentOptions build() {
 
     public TestEnvironmentOptions validateAndBuildWithDefaults() {
       return new TestEnvironmentOptions(
-          workflowClientOptions == null
-              ? WorkflowClientOptions.newBuilder().validateAndBuildWithDefaults()
-              : workflowClientOptions,
-          workerFactoryOptions == null
-              ? WorkerFactoryOptions.newBuilder().validateAndBuildWithDefaults()
-              : workerFactoryOptions,
+          WorkflowClientOptions.newBuilder(workflowClientOptions).validateAndBuildWithDefaults(),
+          WorkerFactoryOptions.newBuilder(workerFactoryOptions).validateAndBuildWithDefaults(),
           metricsScope == null ? new NoopScope() : metricsScope);
     }
   }

src/main/java/io/temporal/testing/TestEnvironmentOptions.java

@@ -92,7 +92,8 @@ public final class Worker implements Suspendable {
 
     this.taskList = taskList;
     this.options = WorkerOptions.newBuilder(options).validateAndBuildWithDefaults();
-    this.factoryOptions = factoryOptions;
+    this.factoryOptions =
+        WorkerFactoryOptions.newBuilder(factoryOptions).validateAndBuildWithDefaults();
     WorkflowServiceStubs service = client.getWorkflowServiceStubs();
     WorkflowClientOptions clientOptions = client.getOptions();
     String namespace = clientOptions.getNamespace();
@@ -105,7 +106,13 @@ public final class Worker implements Suspendable {
             taskList,
             contextPropagators,
             metricsScope);
-    activityWorker = new SyncActivityWorker(service, namespace, taskList, activityOptions);
+    activityWorker =
+        new SyncActivityWorker(
+            service,
+            namespace,
+            taskList,
+            this.options.getTaskListActivitiesPerSecond(),
+            activityOptions);
 
     SingleWorkerOptions workflowOptions =
         toWorkflowOptions(
@@ -128,12 +135,13 @@ public final class Worker implements Suspendable {
             service,
             namespace,
             taskList,
-            factoryOptions.getWorkflowInterceptor(),
+            this.factoryOptions.getWorkflowInterceptor(),
             workflowOptions,
             localActivityOptions,
             this.cache,
             this.stickyTaskListName,
-            Duration.ofSeconds(factoryOptions.getStickyDecisionScheduleToStartTimeoutInSeconds()),
+            Duration.ofSeconds(
+                this.factoryOptions.getWorkflowHostLocalTaskListScheduleToStartTimeoutSeconds()),
             this.threadPoolExecutor);
   }
 
@@ -155,6 +163,7 @@ private static SingleWorkerOptions toActivityOptions(
         .setPollerOptions(
             PollerOptions.newBuilder()
                 .setMaximumPollRatePerSecond(options.getMaxActivitiesPerSecond())
+                .setPollThreadCount(options.getActivityPollThreadCount())
                 .build())
         .setTaskExecutorThreadPoolSize(options.getMaxConcurrentActivityExecutionSize())
         .setMetricsScope(metricsScope.tagged(tags))
@@ -178,7 +187,10 @@ private static SingleWorkerOptions toWorkflowOptions(
     return SingleWorkerOptions.newBuilder()
         .setDataConverter(clientOptions.getDataConverter())
         .setIdentity(clientOptions.getIdentity())
-        .setPollerOptions(PollerOptions.newBuilder().build())
+        .setPollerOptions(
+            PollerOptions.newBuilder()
+                .setPollThreadCount(options.getWorkflowPollThreadCount())
+                .build())
         .setTaskExecutorThreadPoolSize(options.getMaxConcurrentWorkflowTaskExecutionSize())
         .setMetricsScope(metricsScope.tagged(tags))
         .setEnableLoggingInReplay(factoryOptions.isEnableLoggingInReplay())