Merge branch 'main' into workeroptions-builder-isUsingVirtualThreadsOnWorkflowWorker-fix

Author: geraldw-ai

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,11 +10,15 @@
  */
 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.getPollerAutoscalingAutoEnroll()) {
+      pollerAutoscalingAutoEnroll.set(true);
+    }
     if (capabilities.getPollerAutoscaling()) {
       pollerAutoscaling.set(true);
     }
@@ -33,6 +37,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,15 @@ 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);
+
+    // Poller types the user left at their default are auto-enrolled into poller autoscaling at
+    // start() when the namespace advertises the PollerAutoscalingAutoEnroll capability. Eligibility
+    // tracks whether the user called a poller setter (recorded on WorkerOptions.Builder), not the
+    // resolved value, so a defaulted count of 5 is not mistaken for an explicit choice.
+    boolean workflowTaskAutoEnrollEligible = this.options.isWorkflowTaskPollerAutoEnrollEligible();
+    boolean activityTaskAutoEnrollEligible = this.options.isActivityTaskPollerAutoEnrollEligible();
+    boolean nexusTaskAutoEnrollEligible = this.options.isNexusTaskPollerAutoEnrollEligible();
+
     SingleWorkerOptions activityOptions =
         toActivityOptions(
             factoryOptions,
@@ -140,7 +149,8 @@ private static final class TaskSnapshot {
             contextPropagators,
             taggedScope,
             workerInstanceKey,
-            workerControlTaskQueue);
+            workerControlTaskQueue,
+            activityTaskAutoEnrollEligible);
     if (this.options.isLocalActivityWorkerOnly()) {
       activityWorker = null;
     } else {
@@ -174,7 +184,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 +205,8 @@ private static final class TaskSnapshot {
             contextPropagators,
             taggedScope,
             workerInstanceKey,
-            workerControlTaskQueue);
+            workerControlTaskQueue,
+            workflowTaskAutoEnrollEligible);
     SingleWorkerOptions localActivityOptions =
         toLocalActivityOptions(
             factoryOptions,
@@ -901,7 +913,8 @@ private static SingleWorkerOptions toActivityOptions(
       List<ContextPropagator> contextPropagators,
       Scope metricsScope,
       String workerInstanceKey,
-      String workerControlTaskQueue) {
+      String workerControlTaskQueue,
+      boolean autoEnrollEligible) {
     return toSingleWorkerOptions(
             factoryOptions,
             options,
@@ -920,6 +933,7 @@ private static SingleWorkerOptions toActivityOptions(
                         : new PollerBehaviorSimpleMaximum(
                             options.getMaxConcurrentActivityTaskPollers()))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnActivityWorker())
+                .setAutoscalingAutoEnrollEligible(autoEnrollEligible)
                 .build())
         .setMetricsScope(metricsScope)
         .build();
@@ -932,7 +946,8 @@ private static SingleWorkerOptions toNexusOptions(
       List<ContextPropagator> contextPropagators,
       Scope metricsScope,
       String workerInstanceKey,
-      String workerControlTaskQueue) {
+      String workerControlTaskQueue,
+      boolean autoEnrollEligible) {
     return toSingleWorkerOptions(
             factoryOptions,
             options,
@@ -948,6 +963,7 @@ private static SingleWorkerOptions toNexusOptions(
                         : new PollerBehaviorSimpleMaximum(
                             options.getMaxConcurrentNexusTaskPollers()))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
+                .setAutoscalingAutoEnrollEligible(autoEnrollEligible)
                 .build())
         .setMetricsScope(metricsScope)
         .setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
@@ -962,7 +978,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 +1022,7 @@ private static SingleWorkerOptions toWorkflowWorkerOptions(
                         ? pollerBehavior
                         : new PollerBehaviorSimpleMaximum(maxConcurrentWorkflowTaskPollers))
                 .setUsingVirtualThreads(options.isUsingVirtualThreadsOnWorkflowWorker())
+                .setAutoscalingAutoEnrollEligible(autoEnrollEligible)
                 .build())
         .setStickyQueueScheduleToStartTimeout(stickyQueueScheduleToStartTimeout)
         .setStickyTaskQueueDrainTimeout(options.getStickyTaskQueueDrainTimeout())