Auto-enroll pollers into autoscaling on PollerAutoscalingAutoEnroll (Java SDK)

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

@@ -43,7 +43,7 @@ final class ActivityWorker implements SuspendableWorker {
   private final String taskQueue;
   private final SingleWorkerOptions options;
   private final double taskQueueActivitiesPerSecond;
-  private final PollerOptions pollerOptions;
+  private PollerOptions pollerOptions;
   private final Scope workerMetricsScope;
   private final GrpcRetryer grpcRetryer;
   private final GrpcRetryer.GrpcRetryerOptions replyGrpcRetryerOptions;
@@ -83,6 +83,11 @@ public ActivityWorker(
   @Override
   public boolean start() {
     if (handler.isAnyTypeSupported()) {
+      // Auto-enroll into poller autoscaling if the namespace advertises the capability and this
+      // poller type was left at its default. Resolved here (after namespace capabilities are known)
+      // so the poller built below reflects the effective behavior.
+      this.pollerOptions =
+          PollerOptions.maybeEnrollInPollerAutoscaling(pollerOptions, namespaceCapabilities);
       this.pollTaskExecutor =
           new PollTaskExecutor<>(
               namespace,

temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java

@@ -10,12 +10,18 @@
  */
 public final class NamespaceCapabilities {
   private final AtomicBoolean pollerAutoscaling = new AtomicBoolean(false);
+  private final AtomicBoolean pollerAutoscalingAutoEnroll = new AtomicBoolean(false);
   private final AtomicBoolean gracefulPollShutdown = new AtomicBoolean(false);
   private final AtomicBoolean workerHeartbeats = new AtomicBoolean(false);
   private final AtomicBoolean workerCommands = new AtomicBoolean(false);
 
   public void setFromCapabilities(Capabilities capabilities) {
-    if (capabilities.getPollerAutoscaling()) {
+    if (capabilities.getPollerAutoscalingAutoEnroll()) {
+      pollerAutoscalingAutoEnroll.set(true);
+    }
+    // Auto-enroll implies full poller autoscaling support, including scaling down, so it also
+    // enables pollerAutoscaling (which drives serverSupportsAutoscaling in PollScaleReportHandle).
+    if (capabilities.getPollerAutoscaling() || capabilities.getPollerAutoscalingAutoEnroll()) {
       pollerAutoscaling.set(true);
     }
     if (capabilities.getWorkerPollCompleteOnShutdown()) {
@@ -33,6 +39,10 @@ public boolean isPollerAutoscaling() {
     return pollerAutoscaling.get();
   }
 
+  public boolean isPollerAutoscalingAutoEnroll() {
+    return pollerAutoscalingAutoEnroll.get();
+  }
+
   public boolean isGracefulPollShutdown() {
     return gracefulPollShutdown.get();
   }

temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java

@@ -46,7 +46,7 @@ final class NexusWorker implements SuspendableWorker {
   private final String namespace;
   private final String taskQueue;
   private final SingleWorkerOptions options;
-  private final PollerOptions pollerOptions;
+  private PollerOptions pollerOptions;
   private final Scope workerMetricsScope;
   private final DataConverter dataConverter;
   private final GrpcRetryer grpcRetryer;
@@ -114,6 +114,11 @@ public NexusWorker(
   @Override
   public boolean start() {
     if (handler.start()) {
+      // Auto-enroll into poller autoscaling if the namespace advertises the capability and this
+      // poller type was left at its default. Resolved here (after namespace capabilities are known)
+      // so the poller built below reflects the effective behavior.
+      this.pollerOptions =
+          PollerOptions.maybeEnrollInPollerAutoscaling(pollerOptions, namespaceCapabilities);
       this.pollTaskExecutor =
           new PollTaskExecutor<>(
               namespace,

temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java

@@ -3,6 +3,7 @@
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.temporal.worker.tuning.PollerBehavior;
+import io.temporal.worker.tuning.PollerBehaviorAutoscaling;
 import java.time.Duration;
 import java.util.concurrent.ExecutorService;
 import org.slf4j.Logger;
@@ -26,6 +27,27 @@ public static PollerOptions getDefaultInstance() {
     return DEFAULT_INSTANCE;
   }
 
+  /**
+   * If the given options are eligible for poller-autoscaling auto-enrollment (the user left this
+   * poller type at its default) and the namespace advertises the auto-enroll capability, returns a
+   * copy of the options with a default {@link PollerBehaviorAutoscaling} behavior. Otherwise
+   * returns the options unchanged.
+   *
+   * <p>Must only be called before the worker's poller is created (i.e. at worker start), so the
+   * resolved behavior is picked up when the poller is built.
+   */
+  public static PollerOptions maybeEnrollInPollerAutoscaling(
+      PollerOptions options, NamespaceCapabilities namespaceCapabilities) {
+    if (options.isAutoscalingAutoEnrollEligible()
+        && namespaceCapabilities.isPollerAutoscalingAutoEnroll()
+        && !(options.getPollerBehavior() instanceof PollerBehaviorAutoscaling)) {
+      return PollerOptions.newBuilder(options)
+          .setPollerBehavior(new PollerBehaviorAutoscaling())
+          .build();
+    }
+    return options;
+  }
+
   private static final PollerOptions DEFAULT_INSTANCE;
 
   static {
@@ -46,6 +68,7 @@ public static final class Builder {
     private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
     private boolean usingVirtualThreads;
     private ExecutorService pollerTaskExecutorOverride;
+    private boolean autoscalingAutoEnrollEligible;
 
     private Builder() {}
 
@@ -65,6 +88,7 @@ private Builder(PollerOptions options) {
       this.uncaughtExceptionHandler = options.getUncaughtExceptionHandler();
       this.usingVirtualThreads = options.isUsingVirtualThreads();
       this.pollerTaskExecutorOverride = options.getPollerTaskExecutorOverride();
+      this.autoscalingAutoEnrollEligible = options.isAutoscalingAutoEnrollEligible();
     }
 
     /** Defines interval for measuring poll rate. Larger the interval more spiky can be the load. */
@@ -152,6 +176,16 @@ public Builder setPollerTaskExecutorOverride(ExecutorService overrideTaskExecuto
       return this;
     }
 
+    /**
+     * Marks whether this poller type was left at its default (the user set neither a fixed poller
+     * count nor a poller behavior) and is therefore eligible for poller-autoscaling auto-enrollment
+     * when the namespace advertises the capability.
+     */
+    public Builder setAutoscalingAutoEnrollEligible(boolean autoscalingAutoEnrollEligible) {
+      this.autoscalingAutoEnrollEligible = autoscalingAutoEnrollEligible;
+      return this;
+    }
+
     public PollerOptions build() {
       if (uncaughtExceptionHandler == null) {
         uncaughtExceptionHandler =
@@ -180,7 +214,8 @@ public PollerOptions build() {
           uncaughtExceptionHandler,
           pollThreadNamePrefix,
           usingVirtualThreads,
-          pollerTaskExecutorOverride);
+          pollerTaskExecutorOverride,
+          autoscalingAutoEnrollEligible);
     }
   }
 
@@ -198,6 +233,7 @@ public PollerOptions build() {
   private final boolean usingVirtualThreads;
   private final ExecutorService pollerTaskExecutorOverride;
   private final PollerBehavior pollerBehavior;
+  private final boolean autoscalingAutoEnrollEligible;
 
   private PollerOptions(
       int maximumPollRateIntervalMilliseconds,
@@ -211,7 +247,8 @@ private PollerOptions(
       Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
       String pollThreadNamePrefix,
       boolean usingVirtualThreads,
-      ExecutorService pollerTaskExecutorOverride) {
+      ExecutorService pollerTaskExecutorOverride,
+      boolean autoscalingAutoEnrollEligible) {
     this.maximumPollRateIntervalMilliseconds = maximumPollRateIntervalMilliseconds;
     this.maximumPollRatePerSecond = maximumPollRatePerSecond;
     this.backoffCoefficient = backoffCoefficient;
@@ -224,6 +261,7 @@ private PollerOptions(
     this.pollThreadNamePrefix = pollThreadNamePrefix;
     this.usingVirtualThreads = usingVirtualThreads;
     this.pollerTaskExecutorOverride = pollerTaskExecutorOverride;
+    this.autoscalingAutoEnrollEligible = autoscalingAutoEnrollEligible;
   }
 
   public int getMaximumPollRateIntervalMilliseconds() {
@@ -274,6 +312,10 @@ public ExecutorService getPollerTaskExecutorOverride() {
     return pollerTaskExecutorOverride;
   }
 
+  public boolean isAutoscalingAutoEnrollEligible() {
+    return autoscalingAutoEnrollEligible;
+  }
+
   @Override
   public String toString() {
     return "PollerOptions{"

temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java

@@ -48,7 +48,7 @@ final class WorkflowWorker implements SuspendableWorker {
   private final WorkflowExecutorCache cache;
   private final WorkflowTaskHandler handler;
   private final String stickyTaskQueueName;
-  private final PollerOptions pollerOptions;
+  private PollerOptions pollerOptions;
   private final Scope workerMetricsScope;
   private final GrpcRetryer grpcRetryer;
   private final EagerActivityDispatcher eagerActivityDispatcher;
@@ -99,6 +99,11 @@ public WorkflowWorker(
   @Override
   public boolean start() {
     if (handler.isAnyTypeSupported()) {
+      // Auto-enroll into poller autoscaling if the namespace advertises the capability and this
+      // poller type was left at its default. Resolved here (after namespace capabilities are known)
+      // so the poller built below reflects the effective behavior.
+      this.pollerOptions =
+          PollerOptions.maybeEnrollInPollerAutoscaling(pollerOptions, namespaceCapabilities);
       pollTaskExecutor =
           new PollTaskExecutor<>(
               namespace,

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

@@ -132,6 +132,27 @@ private static final class TaskSnapshot {
     Map<String, String> tags =
         new ImmutableMap.Builder<String, String>(1).put(MetricsTag.TASK_QUEUE, taskQueue).build();
     Scope taggedScope = metricsScope.tagged(tags);
+
+    // Determine, per poller type, whether the user left it at its default (set neither a fixed
+    // poller count nor a poller behavior). Eligible types are auto-enrolled into poller autoscaling
+    // at start() when the namespace advertises the PollerAutoscalingAutoEnroll capability. This
+    // must
+    // be read from the raw (pre-defaulting) options, since validateAndBuildWithDefaults() fills in
+    // a
+    // fixed default poller count that would otherwise look like an explicit choice.
+    boolean workflowTaskAutoEnrollEligible =
+        options == null
+            || (options.getWorkflowTaskPollersBehavior() == null
+                && options.getMaxConcurrentWorkflowTaskPollers() == 0);
+    boolean activityTaskAutoEnrollEligible =
+        options == null
+            || (options.getActivityTaskPollersBehavior() == null
+                && options.getMaxConcurrentActivityTaskPollers() == 0);
+    boolean nexusTaskAutoEnrollEligible =
+        options == null
+            || (options.getNexusTaskPollersBehavior() == null
+                && options.getMaxConcurrentNexusTaskPollers() == 0);
+
     SingleWorkerOptions activityOptions =
         toActivityOptions(
             factoryOptions,
@@ -140,7 +161,8 @@ private static final class TaskSnapshot {
             contextPropagators,
             taggedScope,
             workerInstanceKey,
-            workerControlTaskQueue);
+            workerControlTaskQueue,
+            activityTaskAutoEnrollEligible);
     if (this.options.isLocalActivityWorkerOnly()) {
       activityWorker = null;
     } else {
@@ -174,7 +196,8 @@ private static final class TaskSnapshot {
             contextPropagators,
             taggedScope,
             workerInstanceKey,
-            workerControlTaskQueue);
+            workerControlTaskQueue,
+            nexusTaskAutoEnrollEligible);
     SlotSupplier<NexusSlotInfo> nexusSlotSupplier =
         this.options.getWorkerTuner() == null
             ? new FixedSizeSlotSupplier<>(this.options.getMaxConcurrentNexusExecutionSize())
@@ -194,7 +217,8 @@ private static final class TaskSnapshot {
             contextPropagators,
             taggedScope,
             workerInstanceKey,
-            workerControlTaskQueue);
+            workerControlTaskQueue,
+            workflowTaskAutoEnrollEligible);
     SingleWorkerOptions localActivityOptions =
         toLocalActivityOptions(
             factoryOptions,
@@ -901,7 +925,8 @@ private static SingleWorkerOptions toActivityOptions(
       List<ContextPropagator> contextPropagators,
       Scope metricsScope,
       String workerInstanceKey,
-      String workerControlTaskQueue) {
+      String workerControlTaskQueue,
+      boolean autoEnrollEligible) {
     return toSingleWorkerOptions(
             factoryOptions,
             options,
@@ -920,6 +945,7 @@ private static SingleWorkerOptions toActivityOptions(
                         : new PollerBehaviorSimpleMaximum(
                             options.getMaxConcurrentActivityTaskPollers()))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnActivityWorker())
+                .setAutoscalingAutoEnrollEligible(autoEnrollEligible)
                 .build())
         .setMetricsScope(metricsScope)
         .build();
@@ -932,7 +958,8 @@ private static SingleWorkerOptions toNexusOptions(
       List<ContextPropagator> contextPropagators,
       Scope metricsScope,
       String workerInstanceKey,
-      String workerControlTaskQueue) {
+      String workerControlTaskQueue,
+      boolean autoEnrollEligible) {
     return toSingleWorkerOptions(
             factoryOptions,
             options,
@@ -948,6 +975,7 @@ private static SingleWorkerOptions toNexusOptions(
                         : new PollerBehaviorSimpleMaximum(
                             options.getMaxConcurrentNexusTaskPollers()))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
+                .setAutoscalingAutoEnrollEligible(autoEnrollEligible)
                 .build())
         .setMetricsScope(metricsScope)
         .setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
@@ -962,7 +990,8 @@ private static SingleWorkerOptions toWorkflowWorkerOptions(
       List<ContextPropagator> contextPropagators,
       Scope metricsScope,
       String workerInstanceKey,
-      String workerControlTaskQueue) {
+      String workerControlTaskQueue,
+      boolean autoEnrollEligible) {
     Map<String, String> tags =
         new ImmutableMap.Builder<String, String>(1).put(MetricsTag.TASK_QUEUE, taskQueue).build();
 
@@ -1005,6 +1034,7 @@ private static SingleWorkerOptions toWorkflowWorkerOptions(
                         ? pollerBehavior
                         : new PollerBehaviorSimpleMaximum(maxConcurrentWorkflowTaskPollers))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnWorkflowWorker())
+                .setAutoscalingAutoEnrollEligible(autoEnrollEligible)
                 .build())
         .setStickyQueueScheduleToStartTimeout(stickyQueueScheduleToStartTimeout)
         .setStickyTaskQueueDrainTimeout(options.getStickyTaskQueueDrainTimeout())

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

@@ -229,6 +229,10 @@ public Builder setMaxTaskQueueActivitiesPerSecond(double maxTaskQueueActivitiesP
      * value cannot be 1 and will be adjusted to 2 if set to that value.
      *
      * <p>Default is 5, which is chosen if set to zero.
+     *
+     * <p>NOTE: If neither this nor {@link #setWorkflowTaskPollersBehavior} is set and the worker's
+     * namespace is configured to auto-enroll workers into poller autoscaling, the worker will
+     * automatically use poller autoscaling for workflow tasks instead of a fixed number of pollers.
      */
     public Builder setMaxConcurrentWorkflowTaskPollers(int maxConcurrentWorkflowTaskPollers) {
       this.maxConcurrentWorkflowTaskPollers = maxConcurrentWorkflowTaskPollers;
@@ -241,6 +245,10 @@ public Builder setMaxConcurrentWorkflowTaskPollers(int maxConcurrentWorkflowTask
      * tasks from a task queue.
      *
      * <p>Default is 5, which is chosen if set to zero.
+     *
+     * <p>NOTE: If neither this nor {@link #setNexusTaskPollersBehavior} is set and the worker's
+     * namespace is configured to auto-enroll workers into poller autoscaling, the worker will
+     * automatically use poller autoscaling for nexus tasks instead of a fixed number of pollers.
      */
     @Experimental
     public Builder setMaxConcurrentNexusTaskPollers(int maxConcurrentNexusTaskPollers) {
@@ -268,6 +276,10 @@ public Builder setWorkflowPollThreadCount(int workflowPollThreadCount) {
      * `MaxConcurrentActivityExecutionSize` options and still cannot keep up with the request rate.
      *
      * <p>Default is 5, which is chosen if set to zero.
+     *
+     * <p>NOTE: If neither this nor {@link #setActivityTaskPollersBehavior} is set and the worker's
+     * namespace is configured to auto-enroll workers into poller autoscaling, the worker will
+     * automatically use poller autoscaling for activity tasks instead of a fixed number of pollers.
      */
     public Builder setMaxConcurrentActivityTaskPollers(int maxConcurrentActivityTaskPollers) {
       this.maxConcurrentActivityTaskPollers = maxConcurrentActivityTaskPollers;
@@ -505,19 +517,38 @@ public Builder setDeploymentOptions(WorkerDeploymentOptions deploymentOptions) {
      *
      * <p>If the sticky queue is enabled, the poller behavior will be used for the sticky queue as
      * well.
+     *
+     * <p>NOTE: If neither this nor {@link #setMaxConcurrentWorkflowTaskPollers} is set and the
+     * worker's namespace is configured to auto-enroll workers into poller autoscaling, the worker
+     * will automatically use poller autoscaling for workflow tasks instead of a fixed number of
+     * pollers.
      */
     public Builder setWorkflowTaskPollersBehavior(PollerBehavior pollerBehavior) {
       this.workflowTaskPollersBehavior = pollerBehavior;
       return this;
     }
 
-    /** Set the poller behavior for activity task pollers. */
+    /**
+     * Set the poller behavior for activity task pollers.
+     *
+     * <p>NOTE: If neither this nor {@link #setMaxConcurrentActivityTaskPollers} is set and the
+     * worker's namespace is configured to auto-enroll workers into poller autoscaling, the worker
+     * will automatically use poller autoscaling for activity tasks instead of a fixed number of
+     * pollers.
+     */
     public Builder setActivityTaskPollersBehavior(PollerBehavior pollerBehavior) {
       this.activityTaskPollersBehavior = pollerBehavior;
       return this;
     }
 
-    /** Set the poller behavior for nexus task pollers. */
+    /**
+     * Set the poller behavior for nexus task pollers.
+     *
+     * <p>NOTE: If neither this nor {@link #setMaxConcurrentNexusTaskPollers} is set and the
+     * worker's namespace is configured to auto-enroll workers into poller autoscaling, the worker
+     * will automatically use poller autoscaling for nexus tasks instead of a fixed number of
+     * pollers.
+     */
     public Builder setNexusTaskPollersBehavior(PollerBehavior pollerBehavior) {
       this.nexusTaskPollersBehavior = pollerBehavior;
       return this;