Add poller autoscaling options for Springboot (#2554)

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/worker/Worker.java

@@ -560,8 +560,8 @@ private static SingleWorkerOptions toActivityOptions(
             PollerOptions.newBuilder()
                 .setMaximumPollRatePerSecond(options.getMaxWorkerActivitiesPerSecond())
                 .setPollerBehavior(
-                    options.getActivityTaskPollersBehaviour() != null
-                        ? options.getActivityTaskPollersBehaviour()
+                    options.getActivityTaskPollersBehavior() != null
+                        ? options.getActivityTaskPollersBehavior()
                         : new PollerBehaviorSimpleMaximum(
                             options.getMaxConcurrentActivityTaskPollers()))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnActivityWorker())
@@ -580,8 +580,8 @@ private static SingleWorkerOptions toNexusOptions(
         .setPollerOptions(
             PollerOptions.newBuilder()
                 .setPollerBehavior(
-                    options.getNexusTaskPollersBehaviour() != null
-                        ? options.getNexusTaskPollersBehaviour()
+                    options.getNexusTaskPollersBehavior() != null
+                        ? options.getNexusTaskPollersBehavior()
                         : new PollerBehaviorSimpleMaximum(
                             options.getMaxConcurrentNexusTaskPollers()))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
@@ -620,8 +620,8 @@ private static SingleWorkerOptions toWorkflowWorkerOptions(
         .setPollerOptions(
             PollerOptions.newBuilder()
                 .setPollerBehavior(
-                    options.getWorkflowTaskPollersBehaviour() != null
-                        ? options.getWorkflowTaskPollersBehaviour()
+                    options.getWorkflowTaskPollersBehavior() != null
+                        ? options.getWorkflowTaskPollersBehavior()
                         : new PollerBehaviorSimpleMaximum(maxConcurrentWorkflowTaskPollers))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnWorkflowWorker())
                 .build())

temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java

@@ -74,9 +74,9 @@ public static final class Builder {
     private boolean usingVirtualThreadsOnNexusWorker;
     private String identity;
     private WorkerDeploymentOptions deploymentOptions;
-    private PollerBehavior workflowTaskPollersBehaviour;
-    private PollerBehavior activityTaskPollersBehaviour;
-    private PollerBehavior nexusTaskPollersBehaviour;
+    private PollerBehavior workflowTaskPollersBehavior;
+    private PollerBehavior activityTaskPollersBehavior;
+    private PollerBehavior nexusTaskPollersBehavior;
 
     private Builder() {}
 
@@ -109,9 +109,9 @@ private Builder(WorkerOptions o) {
       this.usingVirtualThreadsOnLocalActivityWorker = o.usingVirtualThreadsOnLocalActivityWorker;
       this.usingVirtualThreadsOnNexusWorker = o.usingVirtualThreadsOnNexusWorker;
       this.deploymentOptions = o.deploymentOptions;
-      this.workflowTaskPollersBehaviour = o.workflowTaskPollersBehaviour;
-      this.activityTaskPollersBehaviour = o.activityTaskPollersBehaviour;
-      this.nexusTaskPollersBehaviour = o.nexusTaskPollersBehaviour;
+      this.workflowTaskPollersBehavior = o.workflowTaskPollersBehavior;
+      this.activityTaskPollersBehavior = o.activityTaskPollersBehavior;
+      this.nexusTaskPollersBehavior = o.nexusTaskPollersBehavior;
     }
 
     /**
@@ -500,22 +500,22 @@ public Builder setDeploymentOptions(WorkerDeploymentOptions deploymentOptions) {
      * well.
      */
     @Experimental
-    public Builder setWorkflowTaskPollersBehaviour(PollerBehavior pollerBehavior) {
-      this.workflowTaskPollersBehaviour = pollerBehavior;
+    public Builder setWorkflowTaskPollersBehavior(PollerBehavior pollerBehavior) {
+      this.workflowTaskPollersBehavior = pollerBehavior;
       return this;
     }
 
     /** Set the poller behavior for activity task pollers. */
     @Experimental
-    public Builder setActivityTaskPollersBehaviour(PollerBehavior pollerBehavior) {
-      this.activityTaskPollersBehaviour = pollerBehavior;
+    public Builder setActivityTaskPollersBehavior(PollerBehavior pollerBehavior) {
+      this.activityTaskPollersBehavior = pollerBehavior;
       return this;
     }
 
     /** Set the poller behavior for nexus task pollers. */
     @Experimental
-    public Builder setNexusTaskPollersBehaviour(PollerBehavior pollerBehavior) {
-      this.nexusTaskPollersBehaviour = pollerBehavior;
+    public Builder setNexusTaskPollersBehavior(PollerBehavior pollerBehavior) {
+      this.nexusTaskPollersBehavior = pollerBehavior;
       return this;
     }
 
@@ -546,9 +546,9 @@ public WorkerOptions build() {
           usingVirtualThreadsOnLocalActivityWorker,
           usingVirtualThreadsOnNexusWorker,
           deploymentOptions,
-          workflowTaskPollersBehaviour,
-          activityTaskPollersBehaviour,
-          nexusTaskPollersBehaviour);
+          workflowTaskPollersBehavior,
+          activityTaskPollersBehavior,
+          nexusTaskPollersBehavior);
     }
 
     public WorkerOptions validateAndBuildWithDefaults() {
@@ -608,6 +608,22 @@ public WorkerOptions validateAndBuildWithDefaults() {
       Preconditions.checkState(
           maxConcurrentNexusTaskPollers >= 0, "negative maxConcurrentNexusTaskPollers");
 
+      if (workflowTaskPollersBehavior != null) {
+        Preconditions.checkState(
+            maxConcurrentWorkflowTaskPollers == 0,
+            "workflowTaskPollersBehavior and maxConcurrentWorkflowTaskPollers are mutually exclusive");
+      }
+      if (activityTaskPollersBehavior != null) {
+        Preconditions.checkState(
+            maxConcurrentActivityTaskPollers == 0,
+            "activityTaskPollersBehavior and maxConcurrentActivityTaskPollers are mutually exclusive");
+      }
+      if (nexusTaskPollersBehavior != null) {
+        Preconditions.checkState(
+            maxConcurrentNexusTaskPollers == 0,
+            "nexusTaskPollersBehavior and maxConcurrentNexusTaskPollers are mutually exclusive");
+      }
+
       return new WorkerOptions(
           maxWorkerActivitiesPerSecond,
           maxConcurrentActivityExecutionSize == 0
@@ -658,9 +674,9 @@ public WorkerOptions validateAndBuildWithDefaults() {
           usingVirtualThreadsOnLocalActivityWorker,
           usingVirtualThreadsOnNexusWorker,
           deploymentOptions,
-          workflowTaskPollersBehaviour,
-          activityTaskPollersBehaviour,
-          nexusTaskPollersBehaviour);
+          workflowTaskPollersBehavior,
+          activityTaskPollersBehavior,
+          nexusTaskPollersBehavior);
     }
   }
 
@@ -689,9 +705,9 @@ public WorkerOptions validateAndBuildWithDefaults() {
   private final boolean usingVirtualThreadsOnLocalActivityWorker;
   private final boolean usingVirtualThreadsOnNexusWorker;
   private final WorkerDeploymentOptions deploymentOptions;
-  private PollerBehavior workflowTaskPollersBehaviour;
-  private PollerBehavior activityTaskPollersBehaviour;
-  private PollerBehavior nexusTaskPollersBehaviour;
+  private final PollerBehavior workflowTaskPollersBehavior;
+  private final PollerBehavior activityTaskPollersBehavior;
+  private final PollerBehavior nexusTaskPollersBehavior;
 
   private WorkerOptions(
       double maxWorkerActivitiesPerSecond,
@@ -719,9 +735,9 @@ private WorkerOptions(
       boolean virtualThreadsEnabledOnLocalActivityWorker,
       boolean virtualThreadsEnabledOnNexusWorker,
       WorkerDeploymentOptions deploymentOptions,
-      PollerBehavior workflowTaskPollersBehaviour,
-      PollerBehavior activityTaskPollersBehaviour,
-      PollerBehavior nexusTaskPollersBehaviour) {
+      PollerBehavior workflowTaskPollersBehavior,
+      PollerBehavior activityTaskPollersBehavior,
+      PollerBehavior nexusTaskPollersBehavior) {
     this.maxWorkerActivitiesPerSecond = maxWorkerActivitiesPerSecond;
     this.maxConcurrentActivityExecutionSize = maxConcurrentActivityExecutionSize;
     this.maxConcurrentWorkflowTaskExecutionSize = maxConcurrentWorkflowTaskExecutionSize;
@@ -747,9 +763,9 @@ private WorkerOptions(
     this.usingVirtualThreadsOnLocalActivityWorker = virtualThreadsEnabledOnLocalActivityWorker;
     this.usingVirtualThreadsOnNexusWorker = virtualThreadsEnabledOnNexusWorker;
     this.deploymentOptions = deploymentOptions;
-    this.workflowTaskPollersBehaviour = workflowTaskPollersBehaviour;
-    this.activityTaskPollersBehaviour = activityTaskPollersBehaviour;
-    this.nexusTaskPollersBehaviour = nexusTaskPollersBehaviour;
+    this.workflowTaskPollersBehavior = workflowTaskPollersBehavior;
+    this.activityTaskPollersBehavior = activityTaskPollersBehavior;
+    this.nexusTaskPollersBehavior = nexusTaskPollersBehavior;
   }
 
   public double getMaxWorkerActivitiesPerSecond() {
@@ -870,16 +886,16 @@ public WorkerDeploymentOptions getDeploymentOptions() {
     return deploymentOptions;
   }
 
-  public PollerBehavior getWorkflowTaskPollersBehaviour() {
-    return workflowTaskPollersBehaviour;
+  public PollerBehavior getWorkflowTaskPollersBehavior() {
+    return workflowTaskPollersBehavior;
   }
 
-  public PollerBehavior getActivityTaskPollersBehaviour() {
-    return activityTaskPollersBehaviour;
+  public PollerBehavior getActivityTaskPollersBehavior() {
+    return activityTaskPollersBehavior;
   }
 
-  public PollerBehavior getNexusTaskPollersBehaviour() {
-    return nexusTaskPollersBehaviour;
+  public PollerBehavior getNexusTaskPollersBehavior() {
+    return nexusTaskPollersBehavior;
   }
 
   @Override
@@ -912,9 +928,9 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond
         && usingVirtualThreadsOnLocalActivityWorker == that.usingVirtualThreadsOnLocalActivityWorker
         && usingVirtualThreadsOnNexusWorker == that.usingVirtualThreadsOnNexusWorker
         && Objects.equals(deploymentOptions, that.deploymentOptions)
-        && Objects.equals(workflowTaskPollersBehaviour, that.workflowTaskPollersBehaviour)
-        && Objects.equals(activityTaskPollersBehaviour, that.activityTaskPollersBehaviour)
-        && Objects.equals(nexusTaskPollersBehaviour, that.nexusTaskPollersBehaviour);
+        && Objects.equals(workflowTaskPollersBehavior, that.workflowTaskPollersBehavior)
+        && Objects.equals(activityTaskPollersBehavior, that.activityTaskPollersBehavior)
+        && Objects.equals(nexusTaskPollersBehavior, that.nexusTaskPollersBehavior);
   }
 
   @Override
@@ -945,9 +961,9 @@ public int hashCode() {
         usingVirtualThreadsOnLocalActivityWorker,
         usingVirtualThreadsOnNexusWorker,
         deploymentOptions,
-        workflowTaskPollersBehaviour,
-        activityTaskPollersBehaviour,
-        nexusTaskPollersBehaviour);
+        workflowTaskPollersBehavior,
+        activityTaskPollersBehavior,
+        nexusTaskPollersBehavior);
   }
 
   @Override
@@ -1004,12 +1020,12 @@ public String toString() {
         + usingVirtualThreadsOnNexusWorker
         + ", deploymentOptions="
         + deploymentOptions
-        + ", workflowTaskPollersBehaviour="
-        + workflowTaskPollersBehaviour
-        + ", activityTaskPollersBehaviour="
-        + activityTaskPollersBehaviour
-        + ", nexusTaskPollersBehaviour="
-        + nexusTaskPollersBehaviour
+        + ", workflowTaskPollersBehavior="
+        + workflowTaskPollersBehavior
+        + ", activityTaskPollersBehavior="
+        + activityTaskPollersBehavior
+        + ", nexusTaskPollersBehavior="
+        + nexusTaskPollersBehavior
         + '}';
   }
 }

temporal-sdk/src/main/java/io/temporal/worker/tuning/PollerBehaviorAutoscaling.java

@@ -1,5 +1,7 @@
 package io.temporal.worker.tuning;
 
+import java.util.Objects;
+
 /**
  * A poller behavior that will automatically scale the number of pollers based on feedback from the
  * server. A slot must be available before beginning polling.
@@ -66,4 +68,31 @@ public int getMaxConcurrentTaskPollers() {
   public int getInitialMaxConcurrentTaskPollers() {
     return initialConcurrentTaskPollers;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) return false;
+    PollerBehaviorAutoscaling that = (PollerBehaviorAutoscaling) o;
+    return minConcurrentTaskPollers == that.minConcurrentTaskPollers
+        && maxConcurrentTaskPollers == that.maxConcurrentTaskPollers
+        && initialConcurrentTaskPollers == that.initialConcurrentTaskPollers;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        minConcurrentTaskPollers, maxConcurrentTaskPollers, initialConcurrentTaskPollers);
+  }
+
+  @Override
+  public String toString() {
+    return "PollerBehaviorAutoscaling{"
+        + "minConcurrentTaskPollers="
+        + minConcurrentTaskPollers
+        + ", maxConcurrentTaskPollers="
+        + maxConcurrentTaskPollers
+        + ", initialConcurrentTaskPollers="
+        + initialConcurrentTaskPollers
+        + '}';
+  }
 }

temporal-sdk/src/main/java/io/temporal/worker/tuning/PollerBehaviorSimpleMaximum.java

@@ -1,5 +1,7 @@
 package io.temporal.worker.tuning;
 
+import java.util.Objects;
+
 /**
  * A poller behavior that will attempt to poll as long as a slot is available, up to the provided
  * maximum. Cannot be less than two for workflow tasks, or one for other tasks.
@@ -28,4 +30,24 @@ public PollerBehaviorSimpleMaximum(int maxConcurrentTaskPollers) {
   public int getMaxConcurrentTaskPollers() {
     return maxConcurrentTaskPollers;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (o == null || getClass() != o.getClass()) return false;
+    PollerBehaviorSimpleMaximum that = (PollerBehaviorSimpleMaximum) o;
+    return maxConcurrentTaskPollers == that.maxConcurrentTaskPollers;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(maxConcurrentTaskPollers);
+  }
+
+  @Override
+  public String toString() {
+    return "PollerBehaviorSimpleMaximum{"
+        + "maxConcurrentTaskPollers="
+        + maxConcurrentTaskPollers
+        + '}';
+  }
 }

temporal-sdk/src/test/java/io/temporal/worker/PollerAutoScaleTests.java

@@ -20,8 +20,8 @@ public class PollerAutoScaleTests {
       SDKTestWorkflowRule.newBuilder()
           .setWorkerOptions(
               WorkerOptions.newBuilder()
-                  .setWorkflowTaskPollersBehaviour(new PollerBehaviorAutoscaling(1, 10, 5))
-                  .setActivityTaskPollersBehaviour(new PollerBehaviorAutoscaling(1, 10, 5))
+                  .setWorkflowTaskPollersBehavior(new PollerBehaviorAutoscaling(1, 10, 5))
+                  .setActivityTaskPollersBehavior(new PollerBehaviorAutoscaling(1, 10, 5))
                   .build())
           .setActivityImplementations(new ResourceBasedTunerTests.ActivitiesImpl())
           .setWorkflowTypes(ResourceBasedTunerTests.ResourceTunerWorkflowImpl.class)

temporal-sdk/src/test/java/io/temporal/worker/shutdown/CleanActivityWorkerShutdownTest.java

@@ -55,7 +55,7 @@ public CleanActivityWorkerShutdownTest(PollerBehavior pollerBehaviorAutoscaling)
             .setWorkflowTypes(TestWorkflowImpl.class)
             .setWorkerOptions(
                 WorkerOptions.newBuilder()
-                    .setActivityTaskPollersBehaviour(pollerBehaviorAutoscaling)
+                    .setActivityTaskPollersBehavior(pollerBehaviorAutoscaling)
                     .build())
             .setActivityImplementations(activitiesImpl)
             .build();

temporal-sdk/src/test/java/io/temporal/worker/shutdown/CleanNexusWorkerShutdownTest.java

@@ -60,7 +60,7 @@ public CleanNexusWorkerShutdownTest(PollerBehavior pollerBehaviorAutoscaling) {
             .setWorkerOptions(
                 WorkerOptions.newBuilder()
                     .setLocalActivityWorkerOnly(true)
-                    .setNexusTaskPollersBehaviour(pollerBehaviorAutoscaling)
+                    .setNexusTaskPollersBehavior(pollerBehaviorAutoscaling)
                     .build())
             .setNexusServiceImplementation(nexusServiceImpl)
             .build();

temporal-sdk/src/test/java/io/temporal/worker/shutdown/StickyWorkflowDrainShutdownTest.java

@@ -41,7 +41,7 @@ public StickyWorkflowDrainShutdownTest(PollerBehavior pollerBehaviorAutoscaling)
             .setUseTimeskipping(false)
             .setWorkerOptions(
                 WorkerOptions.newBuilder()
-                    .setWorkflowTaskPollersBehaviour(pollerBehaviorAutoscaling)
+                    .setWorkflowTaskPollersBehavior(pollerBehaviorAutoscaling)
                     .setStickyTaskQueueDrainTimeout(DRAIN_TIME)
                     .build())
             .setWorkflowServiceStubsOptions(

temporal-sdk/src/test/java/io/temporal/workflow/MetricsTest.java

@@ -198,8 +198,8 @@ public void testWorkerMetricsAutoPoller() throws InterruptedException {
 
     WorkerOptions workerOptions =
         WorkerOptions.newBuilder()
-            .setWorkflowTaskPollersBehaviour(new PollerBehaviorAutoscaling(5, 5, 5))
-            .setActivityTaskPollersBehaviour(new PollerBehaviorAutoscaling(5, 5, 5))
+            .setWorkflowTaskPollersBehavior(new PollerBehaviorAutoscaling(5, 5, 5))
+            .setActivityTaskPollersBehavior(new PollerBehaviorAutoscaling(5, 5, 5))
             .build();
     Worker worker = testEnvironment.newWorker(TASK_QUEUE, workerOptions);
     worker.registerWorkflowImplementationTypes(