From 7c35a150615595a037da3edf9d9d85abc3de4299 Mon Sep 17 00:00:00 2001 From: Veeral Patel Date: Wed, 15 Jul 2026 14:16:52 -0700 Subject: [PATCH 1/4] Auto-enroll pollers into autoscaling on PollerAutoscalingAutoEnroll When a namespace advertises the PollerAutoscalingAutoEnroll capability, automatically switch a poller type to poller autoscaling if the user left it at its default (set neither MaxConcurrentTaskPollers nor a TaskPollerBehavior). Explicitly configured pollers are untouched. Auto-enroll implies full autoscaling support, so it also enables serverSupportsAutoscaling (scale-down). Applies to workflow, activity, and nexus pollers. The decision is made at worker start(), after namespace capabilities are loaded, by rebuilding each dormant worker's PollerOptions before its poller is created. Per-poller-type eligibility is captured at Worker construction from the raw (pre-defaulting) WorkerOptions and threaded through PollerOptions. Also bumps the temporal/api proto submodule to pick up the poller_autoscaling_auto_enroll namespace capability (api #803). --- .../internal/worker/ActivityWorker.java | 7 +- .../worker/NamespaceCapabilities.java | 12 +- .../temporal/internal/worker/NexusWorker.java | 7 +- .../internal/worker/PollerOptions.java | 46 +++++- .../internal/worker/WorkflowWorker.java | 7 +- .../main/java/io/temporal/worker/Worker.java | 42 +++++- .../io/temporal/worker/WorkerOptions.java | 35 ++++- .../PollerAutoscalingAutoEnrollTest.java | 113 ++++++++++++++ ...WorkerPollerAutoEnrollEligibilityTest.java | 138 ++++++++++++++++++ temporal-serviceclient/src/main/proto | 2 +- 10 files changed, 394 insertions(+), 15 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java index 6c86fc4472..ff528d46b3 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java +++ b/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, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java index ed4ac3935f..158e46e257 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java +++ b/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(); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java index 1fd9cf9148..33416a807b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java +++ b/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, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java index 1765c5d1cd..1245907dda 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/PollerOptions.java +++ b/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. + * + *

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{" diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java index 0de3ffaf6b..b86dfea6df 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java +++ b/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, diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 5ba7613865..32191b34a5 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -132,6 +132,27 @@ private static final class TaskSnapshot { Map tags = new ImmutableMap.Builder(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 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 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 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 contextPropagators, Scope metricsScope, String workerInstanceKey, - String workerControlTaskQueue) { + String workerControlTaskQueue, + boolean autoEnrollEligible) { Map tags = new ImmutableMap.Builder(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()) diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java index 4e351cb74d..d181a75eaf 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/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. * *

Default is 5, which is chosen if set to zero. + * + *

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. * *

Default is 5, which is chosen if set to zero. + * + *

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. * *

Default is 5, which is chosen if set to zero. + * + *

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) { * *

If the sticky queue is enabled, the poller behavior will be used for the sticky queue as * well. + * + *

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. + * + *

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. + * + *

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; diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java new file mode 100644 index 0000000000..556f722fa4 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java @@ -0,0 +1,113 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import io.temporal.api.namespace.v1.NamespaceInfo.Capabilities; +import io.temporal.worker.tuning.PollerBehaviorAutoscaling; +import io.temporal.worker.tuning.PollerBehaviorSimpleMaximum; +import org.junit.Test; + +/** + * Tests for poller-autoscaling auto-enrollment: when a namespace advertises the + * PollerAutoscalingAutoEnroll capability, poller types left at their default are switched to poller + * autoscaling, while explicitly-configured poller types are left untouched. + */ +public class PollerAutoscalingAutoEnrollTest { + + private static NamespaceCapabilities capabilities(boolean autoEnroll, boolean pollerAutoscaling) { + NamespaceCapabilities caps = new NamespaceCapabilities(); + caps.setFromCapabilities( + Capabilities.newBuilder() + .setPollerAutoscalingAutoEnroll(autoEnroll) + .setPollerAutoscaling(pollerAutoscaling) + .build()); + return caps; + } + + private static PollerOptions eligibleFixedPollerOptions() { + return PollerOptions.newBuilder() + .setPollerBehavior(new PollerBehaviorSimpleMaximum(5)) + .setAutoscalingAutoEnrollEligible(true) + .build(); + } + + @Test + public void autoEnrollCapabilityImpliesPollerAutoscaling() { + // Auto-enroll implies full autoscaling support (including scale-down), so it also enables the + // pollerAutoscaling flag that PollScaleReportHandle reads as serverSupportsAutoscaling. + NamespaceCapabilities caps = capabilities(true, false); + assertTrue(caps.isPollerAutoscalingAutoEnroll()); + assertTrue(caps.isPollerAutoscaling()); + } + + @Test + public void pollerAutoscalingWithoutAutoEnrollDoesNotImplyAutoEnroll() { + NamespaceCapabilities caps = capabilities(false, true); + assertFalse(caps.isPollerAutoscalingAutoEnroll()); + assertTrue(caps.isPollerAutoscaling()); + } + + @Test + public void noCapabilitiesLeavesEverythingDisabled() { + NamespaceCapabilities caps = capabilities(false, false); + assertFalse(caps.isPollerAutoscalingAutoEnroll()); + assertFalse(caps.isPollerAutoscaling()); + } + + @Test + public void eligibleDefaultIsEnrolledWhenCapabilityAdvertised() { + PollerOptions resolved = + PollerOptions.maybeEnrollInPollerAutoscaling( + eligibleFixedPollerOptions(), capabilities(true, false)); + assertTrue( + "defaulted poller type should switch to autoscaling", + resolved.getPollerBehavior() instanceof PollerBehaviorAutoscaling); + // The enrolled behavior uses the PollerBehaviorAutoscaling defaults. + assertEquals(new PollerBehaviorAutoscaling(), resolved.getPollerBehavior()); + } + + @Test + public void eligibleDefaultIsNotEnrolledWhenCapabilityAbsent() { + PollerOptions options = eligibleFixedPollerOptions(); + PollerOptions resolved = + PollerOptions.maybeEnrollInPollerAutoscaling(options, capabilities(false, false)); + assertSame("without the capability the options are unchanged", options, resolved); + assertTrue(resolved.getPollerBehavior() instanceof PollerBehaviorSimpleMaximum); + } + + @Test + public void explicitlyConfiguredPollerIsNotEnrolled() { + // A poller type the user configured explicitly is not eligible and must not be switched, even + // when the namespace advertises auto-enroll. + PollerOptions options = + PollerOptions.newBuilder() + .setPollerBehavior(new PollerBehaviorSimpleMaximum(3)) + .setAutoscalingAutoEnrollEligible(false) + .build(); + PollerOptions resolved = + PollerOptions.maybeEnrollInPollerAutoscaling(options, capabilities(true, false)); + assertSame("explicitly configured poller is untouched", options, resolved); + assertEquals( + 3, + ((PollerBehaviorSimpleMaximum) resolved.getPollerBehavior()).getMaxConcurrentTaskPollers()); + } + + @Test + public void alreadyAutoscalingPollerIsLeftUnchanged() { + // An eligible poller that already uses autoscaling (e.g. a user-set autoscaling behavior on a + // type otherwise treated as eligible) is returned unchanged rather than rebuilt. + PollerBehaviorAutoscaling behavior = new PollerBehaviorAutoscaling(2, 20, 4); + PollerOptions options = + PollerOptions.newBuilder() + .setPollerBehavior(behavior) + .setAutoscalingAutoEnrollEligible(true) + .build(); + PollerOptions resolved = + PollerOptions.maybeEnrollInPollerAutoscaling(options, capabilities(true, false)); + assertSame(options, resolved); + assertSame(behavior, resolved.getPollerBehavior()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java new file mode 100644 index 0000000000..949491d100 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java @@ -0,0 +1,138 @@ +package io.temporal.worker; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import com.uber.m3.tally.Scope; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.internal.sync.WorkflowThreadExecutor; +import io.temporal.internal.worker.NamespaceCapabilities; +import io.temporal.internal.worker.WorkflowExecutorCache; +import io.temporal.internal.worker.WorkflowRunLockManager; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.tuning.PollerBehaviorSimpleMaximum; +import java.util.Collections; +import org.junit.Test; + +/** + * Verifies that {@link Worker} derives per-poller-type auto-enroll eligibility from the raw + * (pre-defaulting) {@link WorkerOptions} and threads it into each internal worker's poller options. + * + *

This guards the subtlest part of poller-autoscaling auto-enrollment: eligibility must be read + * before {@code validateAndBuildWithDefaults()} fills in a fixed default poller count. If it were + * read from the already-defaulted options, every default worker would look explicitly configured + * and auto-enroll would silently never happen. + */ +public class WorkerPollerAutoEnrollEligibilityTest { + + private Worker buildWorker(WorkerOptions options) { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.newBuilder().build()); + WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = + mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + + WorkflowClient client = mock(WorkflowClient.class); + when(client.getWorkflowServiceStubs()).thenReturn(service); + when(client.getOptions()) + .thenReturn( + WorkflowClientOptions.newBuilder() + .setNamespace("test-ns") + .setIdentity("test-worker") + .validateAndBuildWithDefaults()); + + Scope metricsScope = new NoopScope(); + WorkflowRunLockManager runLocks = new WorkflowRunLockManager(); + WorkflowExecutorCache cache = new WorkflowExecutorCache(10, runLocks, metricsScope); + WorkflowThreadExecutor wfThreadExecutor = mock(WorkflowThreadExecutor.class); + + return new Worker( + client, + "test-task-queue", + WorkerFactoryOptions.newBuilder().build(), + options, + metricsScope, + runLocks, + cache, + true, + wfThreadExecutor, + Collections.emptyList(), + Collections.emptyList(), + "test-worker-group", + new NamespaceCapabilities()); + } + + private boolean workflowEligible(Worker worker) { + return worker.workflowWorker.getWorkflowPollerOptions().isAutoscalingAutoEnrollEligible(); + } + + private boolean activityEligible(Worker worker) { + return worker.activityWorker.getPollerOptions().isAutoscalingAutoEnrollEligible(); + } + + private boolean nexusEligible(Worker worker) { + return worker.nexusWorker.getPollerOptions().isAutoscalingAutoEnrollEligible(); + } + + @Test + public void defaultOptionsMakeEveryPollerTypeEligible() { + Worker worker = buildWorker(WorkerOptions.newBuilder().build()); + assertTrue(workflowEligible(worker)); + assertTrue(activityEligible(worker)); + assertTrue(nexusEligible(worker)); + } + + @Test + public void nullOptionsMakeEveryPollerTypeEligible() { + // WorkerFactory.newWorker(taskQueue) passes null options through to the Worker constructor. + Worker worker = buildWorker(null); + assertTrue(workflowEligible(worker)); + assertTrue(activityEligible(worker)); + assertTrue(nexusEligible(worker)); + } + + @Test + public void explicitMaxConcurrentPollersMakeAllTypesIneligible() { + Worker worker = + buildWorker( + WorkerOptions.newBuilder() + .setMaxConcurrentWorkflowTaskPollers(4) + .setMaxConcurrentActivityTaskPollers(3) + .setMaxConcurrentNexusTaskPollers(2) + .build()); + assertFalse(workflowEligible(worker)); + assertFalse(activityEligible(worker)); + assertFalse(nexusEligible(worker)); + } + + @Test + public void explicitMaxConcurrentPollersOnOneTypeLeavesOthersEligible() { + Worker worker = + buildWorker(WorkerOptions.newBuilder().setMaxConcurrentWorkflowTaskPollers(4).build()); + assertFalse(workflowEligible(worker)); + assertTrue(activityEligible(worker)); + assertTrue(nexusEligible(worker)); + } + + @Test + public void explicitPollerBehaviorMakesOnlyThatTypeIneligible() { + Worker worker = + buildWorker( + WorkerOptions.newBuilder() + .setActivityTaskPollersBehavior(new PollerBehaviorSimpleMaximum(3)) + .build()); + // Only the activity poller was configured explicitly; workflow and nexus stay eligible. + assertFalse(activityEligible(worker)); + assertTrue(workflowEligible(worker)); + assertTrue(nexusEligible(worker)); + } +} diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index d2fc34ab84..852ee3b339 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit d2fc34ab844603f50e41365f46c7fb82bdedffe6 +Subproject commit 852ee3b339f9f1efe3503fe96f0351d361396daa From afcf607afa2e6a167bf0e7a9e2f46f88bbcbbda5 Mon Sep 17 00:00:00 2001 From: Veeral Patel Date: Tue, 21 Jul 2026 14:40:07 -0500 Subject: [PATCH 2/4] Address review: track poller-setter intent for auto-enroll eligibility - Determine auto-enroll eligibility from whether the user actually called a poller setter (MaxConcurrentTaskPollers or TaskPollerBehavior), tracked on WorkerOptions.Builder and carried through build/validate/copy, instead of inferring from the resolved option values. This fixes the case where options from getDefaultInstance()/validateAndBuildWithDefaults() carry the numeric default (5) and would wrongly look explicitly configured. - Worker reads the new WorkerOptions eligibility getters instead of the raw pre-defaulting options. - Reword the NamespaceCapabilities comment to explain why auto-enroll also enables pollerAutoscaling (serverSupportsAutoscaling in PollScaleReportHandle), and fix the mangled comment in Worker. - Expand tests: eligibility from getDefaultInstance()/validateAndBuildWithDefaults() and copies stays eligible; an explicit count equal to the numeric default is ineligible. Add a full start-path E2E test asserting workflow/activity/nexus pollers become autoscaling when the namespace advertises the capability. --- .../worker/NamespaceCapabilities.java | 6 +- .../main/java/io/temporal/worker/Worker.java | 26 +-- .../io/temporal/worker/WorkerOptions.java | 66 +++++++- ...WorkerPollerAutoEnrollEligibilityTest.java | 60 ++++++- .../WorkerPollerAutoEnrollStartupTest.java | 157 ++++++++++++++++++ 5 files changed, 285 insertions(+), 30 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java index 158e46e257..09ec29c02c 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java @@ -19,8 +19,10 @@ public void setFromCapabilities(Capabilities capabilities) { 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). + // The auto-enroll capability implies the server speaks the full poller-scaling protocol, so + // also enable pollerAutoscaling: it is the flag AsyncPoller passes to PollScaleReportHandle as + // serverSupportsAutoscaling, which lets the poller count follow the server's scaling + // suggestions. if (capabilities.getPollerAutoscaling() || capabilities.getPollerAutoscalingAutoEnroll()) { pollerAutoscaling.set(true); } diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 32191b34a5..93a26def2b 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -133,25 +133,13 @@ private static final class TaskSnapshot { new ImmutableMap.Builder(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); + // 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( diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java index d181a75eaf..99a596514b 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java @@ -79,6 +79,13 @@ public static final class Builder { private PollerBehavior nexusTaskPollersBehavior; private boolean allowActivityHeartbeatDuringShutdown; private PreferredVersionProvider preferredVersionProvider; + // Track whether the user explicitly configured the pollers for a task type (called either the + // max-concurrent-pollers or the poller-behavior setter). A type left unconfigured is eligible + // for poller-autoscaling auto-enrollment. This must reflect the user's intent, not the resolved + // value, since defaulting fills in a non-zero count that would otherwise look explicit. + private boolean workflowTaskPollersConfigured; + private boolean activityTaskPollersConfigured; + private boolean nexusTaskPollersConfigured; private Builder() {} @@ -116,6 +123,9 @@ private Builder(WorkerOptions o) { this.nexusTaskPollersBehavior = o.nexusTaskPollersBehavior; this.allowActivityHeartbeatDuringShutdown = o.allowActivityHeartbeatDuringShutdown; this.preferredVersionProvider = o.preferredVersionProvider; + this.workflowTaskPollersConfigured = o.workflowTaskPollersConfigured; + this.activityTaskPollersConfigured = o.activityTaskPollersConfigured; + this.nexusTaskPollersConfigured = o.nexusTaskPollersConfigured; } /** @@ -236,6 +246,7 @@ public Builder setMaxTaskQueueActivitiesPerSecond(double maxTaskQueueActivitiesP */ public Builder setMaxConcurrentWorkflowTaskPollers(int maxConcurrentWorkflowTaskPollers) { this.maxConcurrentWorkflowTaskPollers = maxConcurrentWorkflowTaskPollers; + this.workflowTaskPollersConfigured = true; return this; } @@ -253,6 +264,7 @@ public Builder setMaxConcurrentWorkflowTaskPollers(int maxConcurrentWorkflowTask @Experimental public Builder setMaxConcurrentNexusTaskPollers(int maxConcurrentNexusTaskPollers) { this.maxConcurrentNexusTaskPollers = maxConcurrentNexusTaskPollers; + this.nexusTaskPollersConfigured = true; return this; } @@ -283,6 +295,7 @@ public Builder setWorkflowPollThreadCount(int workflowPollThreadCount) { */ public Builder setMaxConcurrentActivityTaskPollers(int maxConcurrentActivityTaskPollers) { this.maxConcurrentActivityTaskPollers = maxConcurrentActivityTaskPollers; + this.activityTaskPollersConfigured = true; return this; } @@ -525,6 +538,7 @@ public Builder setDeploymentOptions(WorkerDeploymentOptions deploymentOptions) { */ public Builder setWorkflowTaskPollersBehavior(PollerBehavior pollerBehavior) { this.workflowTaskPollersBehavior = pollerBehavior; + this.workflowTaskPollersConfigured = true; return this; } @@ -538,6 +552,7 @@ public Builder setWorkflowTaskPollersBehavior(PollerBehavior pollerBehavior) { */ public Builder setActivityTaskPollersBehavior(PollerBehavior pollerBehavior) { this.activityTaskPollersBehavior = pollerBehavior; + this.activityTaskPollersConfigured = true; return this; } @@ -551,6 +566,7 @@ public Builder setActivityTaskPollersBehavior(PollerBehavior pollerBehavior) { */ public Builder setNexusTaskPollersBehavior(PollerBehavior pollerBehavior) { this.nexusTaskPollersBehavior = pollerBehavior; + this.nexusTaskPollersConfigured = true; return this; } @@ -620,7 +636,10 @@ public WorkerOptions build() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, - preferredVersionProvider); + preferredVersionProvider, + workflowTaskPollersConfigured, + activityTaskPollersConfigured, + nexusTaskPollersConfigured); } public WorkerOptions validateAndBuildWithDefaults() { @@ -754,7 +773,10 @@ public WorkerOptions validateAndBuildWithDefaults() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, - preferredVersionProvider); + preferredVersionProvider, + workflowTaskPollersConfigured, + activityTaskPollersConfigured, + nexusTaskPollersConfigured); } } @@ -788,6 +810,9 @@ public WorkerOptions validateAndBuildWithDefaults() { private final PollerBehavior nexusTaskPollersBehavior; private final boolean allowActivityHeartbeatDuringShutdown; private final PreferredVersionProvider preferredVersionProvider; + private final boolean workflowTaskPollersConfigured; + private final boolean activityTaskPollersConfigured; + private final boolean nexusTaskPollersConfigured; private WorkerOptions( double maxWorkerActivitiesPerSecond, @@ -819,7 +844,10 @@ private WorkerOptions( PollerBehavior activityTaskPollersBehavior, PollerBehavior nexusTaskPollersBehavior, boolean allowActivityHeartbeatDuringShutdown, - PreferredVersionProvider preferredVersionProvider) { + PreferredVersionProvider preferredVersionProvider, + boolean workflowTaskPollersConfigured, + boolean activityTaskPollersConfigured, + boolean nexusTaskPollersConfigured) { this.maxWorkerActivitiesPerSecond = maxWorkerActivitiesPerSecond; this.maxConcurrentActivityExecutionSize = maxConcurrentActivityExecutionSize; this.maxConcurrentWorkflowTaskExecutionSize = maxConcurrentWorkflowTaskExecutionSize; @@ -850,6 +878,38 @@ private WorkerOptions( this.nexusTaskPollersBehavior = nexusTaskPollersBehavior; this.allowActivityHeartbeatDuringShutdown = allowActivityHeartbeatDuringShutdown; this.preferredVersionProvider = preferredVersionProvider; + this.workflowTaskPollersConfigured = workflowTaskPollersConfigured; + this.activityTaskPollersConfigured = activityTaskPollersConfigured; + this.nexusTaskPollersConfigured = nexusTaskPollersConfigured; + } + + /** + * Whether the workflow task pollers were left at their default (the user called neither {@link + * Builder#setMaxConcurrentWorkflowTaskPollers} nor {@link + * Builder#setWorkflowTaskPollersBehavior}), which makes them eligible for poller-autoscaling + * auto-enrollment. + */ + boolean isWorkflowTaskPollerAutoEnrollEligible() { + return !workflowTaskPollersConfigured; + } + + /** + * Whether the activity task pollers were left at their default (the user called neither {@link + * Builder#setMaxConcurrentActivityTaskPollers} nor {@link + * Builder#setActivityTaskPollersBehavior}), which makes them eligible for poller-autoscaling + * auto-enrollment. + */ + boolean isActivityTaskPollerAutoEnrollEligible() { + return !activityTaskPollersConfigured; + } + + /** + * Whether the nexus task pollers were left at their default (the user called neither {@link + * Builder#setMaxConcurrentNexusTaskPollers} nor {@link Builder#setNexusTaskPollersBehavior}), + * which makes them eligible for poller-autoscaling auto-enrollment. + */ + boolean isNexusTaskPollerAutoEnrollEligible() { + return !nexusTaskPollersConfigured; } public double getMaxWorkerActivitiesPerSecond() { diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java index 949491d100..a72f4bd746 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java @@ -22,13 +22,14 @@ import org.junit.Test; /** - * Verifies that {@link Worker} derives per-poller-type auto-enroll eligibility from the raw - * (pre-defaulting) {@link WorkerOptions} and threads it into each internal worker's poller options. + * Verifies that {@link Worker} derives per-poller-type auto-enroll eligibility from whether the + * user called a poller setter (tracked on {@link WorkerOptions.Builder}) and threads it into each + * internal worker's poller options. * - *

This guards the subtlest part of poller-autoscaling auto-enrollment: eligibility must be read - * before {@code validateAndBuildWithDefaults()} fills in a fixed default poller count. If it were - * read from the already-defaulted options, every default worker would look explicitly configured - * and auto-enroll would silently never happen. + *

This guards the subtlest part of poller-autoscaling auto-enrollment: eligibility must reflect + * the user's intent, not the resolved value. Options from {@code getDefaultInstance()} or {@code + * validateAndBuildWithDefaults()} carry the numeric default poller count (5) yet must stay + * eligible, while an explicit count of 5 must not. Value-based inference cannot tell these apart. */ public class WorkerPollerAutoEnrollEligibilityTest { @@ -123,6 +124,53 @@ public void explicitMaxConcurrentPollersOnOneTypeLeavesOthersEligible() { assertTrue(nexusEligible(worker)); } + @Test + public void defaultInstanceMakesEveryPollerTypeEligible() { + // getDefaultInstance() carries the numeric default poller count (5), but the user never called + // a + // setter, so all types remain eligible. + Worker worker = buildWorker(WorkerOptions.getDefaultInstance()); + assertTrue(workflowEligible(worker)); + assertTrue(activityEligible(worker)); + assertTrue(nexusEligible(worker)); + } + + @Test + public void validateAndBuildWithDefaultsMakesEveryPollerTypeEligible() { + Worker worker = buildWorker(WorkerOptions.newBuilder().validateAndBuildWithDefaults()); + assertTrue(workflowEligible(worker)); + assertTrue(activityEligible(worker)); + assertTrue(nexusEligible(worker)); + } + + @Test + public void copyOfDefaultInstanceIsEligible() { + // Provenance must survive newBuilder(options), for both build paths. + Worker fromBuild = + buildWorker(WorkerOptions.newBuilder(WorkerOptions.getDefaultInstance()).build()); + assertTrue(workflowEligible(fromBuild)); + assertTrue(activityEligible(fromBuild)); + assertTrue(nexusEligible(fromBuild)); + + Worker fromValidate = + buildWorker( + WorkerOptions.newBuilder(WorkerOptions.getDefaultInstance()) + .validateAndBuildWithDefaults()); + assertTrue(workflowEligible(fromValidate)); + assertTrue(activityEligible(fromValidate)); + assertTrue(nexusEligible(fromValidate)); + } + + @Test + public void explicitCountEqualToNumericDefaultIsIneligible() { + // Explicitly setting the count to its numeric default (5) still counts as "configured". + Worker worker = + buildWorker(WorkerOptions.newBuilder().setMaxConcurrentWorkflowTaskPollers(5).build()); + assertFalse(workflowEligible(worker)); + assertTrue(activityEligible(worker)); + assertTrue(nexusEligible(worker)); + } + @Test public void explicitPollerBehaviorMakesOnlyThatTypeIneligible() { Worker worker = diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java new file mode 100644 index 0000000000..00ea0d69be --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollStartupTest.java @@ -0,0 +1,157 @@ +package io.temporal.worker; + +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.SettableFuture; +import com.uber.m3.tally.NoopScope; +import com.uber.m3.tally.Scope; +import io.nexusrpc.handler.OperationHandler; +import io.nexusrpc.handler.OperationImpl; +import io.nexusrpc.handler.ServiceImpl; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import io.temporal.api.namespace.v1.NamespaceInfo.Capabilities; +import io.temporal.api.workflowservice.v1.GetSystemInfoResponse; +import io.temporal.api.workflowservice.v1.ShutdownWorkerRequest; +import io.temporal.api.workflowservice.v1.ShutdownWorkerResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.internal.sync.WorkflowThreadExecutor; +import io.temporal.internal.worker.NamespaceCapabilities; +import io.temporal.internal.worker.ShutdownManager; +import io.temporal.internal.worker.WorkflowExecutorCache; +import io.temporal.internal.worker.WorkflowRunLockManager; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.tuning.PollerBehaviorAutoscaling; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import io.temporal.workflow.shared.TestNexusServices; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import org.junit.Test; + +/** + * Full start-path wiring test for poller-autoscaling auto-enrollment: when the namespace advertises + * the capability, starting a worker with default options switches the workflow, activity, and nexus + * pollers' effective behavior to {@link PollerBehaviorAutoscaling}. + */ +public class WorkerPollerAutoEnrollStartupTest { + + @WorkflowInterface + public interface DemoWorkflow { + @WorkflowMethod + void run(); + } + + public static class DemoWorkflowImpl implements DemoWorkflow { + @Override + public void run() {} + } + + @ActivityInterface + public interface DemoActivity { + @ActivityMethod + void doThing(); + } + + public static class DemoActivityImpl implements DemoActivity { + @Override + public void doThing() {} + } + + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public static class DemoNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return OperationHandler.sync((ctx, details, name) -> "Hello " + name); + } + } + + @Test + public void autoEnrollAtStartupSwitchesPollersToAutoscaling() throws Exception { + WorkflowServiceStubs service = mock(WorkflowServiceStubs.class); + when(service.getServerCapabilities()) + .thenReturn(() -> GetSystemInfoResponse.Capabilities.newBuilder().build()); + + // Async pollers poll via futureStub().withOption(...).pollXxxTaskQueue(...). Return futures + // that + // never complete so the started poller threads park harmlessly until shutdown cancels them. + WorkflowServiceGrpc.WorkflowServiceFutureStub futureStub = + mock(WorkflowServiceGrpc.WorkflowServiceFutureStub.class); + when(service.futureStub()).thenReturn(futureStub); + when(futureStub.withOption(any(), any())).thenReturn(futureStub); + when(futureStub.pollWorkflowTaskQueue(any())).thenReturn(SettableFuture.create()); + when(futureStub.pollActivityTaskQueue(any())).thenReturn(SettableFuture.create()); + when(futureStub.pollNexusTaskQueue(any())).thenReturn(SettableFuture.create()); + when(futureStub.shutdownWorker(any(ShutdownWorkerRequest.class))) + .thenReturn(Futures.immediateFuture(ShutdownWorkerResponse.newBuilder().build())); + + WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub = + mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + + WorkflowClient client = mock(WorkflowClient.class); + when(client.getWorkflowServiceStubs()).thenReturn(service); + when(client.getOptions()) + .thenReturn( + WorkflowClientOptions.newBuilder() + .setNamespace("test-ns") + .setIdentity("test-worker") + .validateAndBuildWithDefaults()); + + // Namespace advertises the auto-enroll capability. + NamespaceCapabilities capabilities = new NamespaceCapabilities(); + capabilities.setFromCapabilities( + Capabilities.newBuilder().setPollerAutoscalingAutoEnroll(true).build()); + + Scope metricsScope = new NoopScope(); + WorkflowRunLockManager runLocks = new WorkflowRunLockManager(); + WorkflowExecutorCache cache = new WorkflowExecutorCache(10, runLocks, metricsScope); + WorkflowThreadExecutor wfThreadExecutor = mock(WorkflowThreadExecutor.class); + + Worker worker = + new Worker( + client, + "test-task-queue", + WorkerFactoryOptions.newBuilder().build(), + WorkerOptions.newBuilder().build(), + metricsScope, + runLocks, + cache, + true, + wfThreadExecutor, + Collections.emptyList(), + Collections.emptyList(), + "test-worker-group", + capabilities); + + // Register all three task types so each worker starts its poller. + worker.registerWorkflowImplementationTypes(DemoWorkflowImpl.class); + worker.registerActivitiesImplementations(new DemoActivityImpl()); + worker.registerNexusServiceImplementation(new DemoNexusServiceImpl()); + + worker.start(); + try { + assertTrue( + "workflow pollers should be autoscaling", + worker.workflowWorker.getWorkflowPollerOptions().getPollerBehavior() + instanceof PollerBehaviorAutoscaling); + assertTrue( + "activity pollers should be autoscaling", + worker.activityWorker.getPollerOptions().getPollerBehavior() + instanceof PollerBehaviorAutoscaling); + assertTrue( + "nexus pollers should be autoscaling", + worker.nexusWorker.getPollerOptions().getPollerBehavior() + instanceof PollerBehaviorAutoscaling); + } finally { + worker.shutdown(new ShutdownManager(), true).get(5, TimeUnit.SECONDS); + } + } +} From 24fd3679ccc2ab5272f84289072d3a4f2b59c9eb Mon Sep 17 00:00:00 2001 From: Veeral Patel Date: Tue, 21 Jul 2026 14:54:14 -0500 Subject: [PATCH 3/4] Address review: don't couple auto-enroll to the pollerAutoscaling flag The auto-enroll capability no longer forces NamespaceCapabilities.pollerAutoscaling on. The server advertises pollerAutoscaling independently (and unconditionally), so the pre-existing getPollerAutoscaling() read already enables scale-down for auto-enrolled pollers; the extra coupling was redundant and conflated the scale-down flag with the enrollment decision. setFromCapabilities reverts to reading each capability separately. The auto-enroll field/getter stay, since the enrollment decision still needs them and is independent of isPollerAutoscaling(). Replaced the obsolete implication test with one asserting the two capabilities are independent. --- .../temporal/internal/worker/NamespaceCapabilities.java | 6 +----- .../internal/worker/PollerAutoscalingAutoEnrollTest.java | 9 +++++---- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java index 09ec29c02c..4bddd45d9e 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/NamespaceCapabilities.java @@ -19,11 +19,7 @@ public void setFromCapabilities(Capabilities capabilities) { if (capabilities.getPollerAutoscalingAutoEnroll()) { pollerAutoscalingAutoEnroll.set(true); } - // The auto-enroll capability implies the server speaks the full poller-scaling protocol, so - // also enable pollerAutoscaling: it is the flag AsyncPoller passes to PollScaleReportHandle as - // serverSupportsAutoscaling, which lets the poller count follow the server's scaling - // suggestions. - if (capabilities.getPollerAutoscaling() || capabilities.getPollerAutoscalingAutoEnroll()) { + if (capabilities.getPollerAutoscaling()) { pollerAutoscaling.set(true); } if (capabilities.getWorkerPollCompleteOnShutdown()) { diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java index 556f722fa4..5d0df9cc7b 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java @@ -35,12 +35,13 @@ private static PollerOptions eligibleFixedPollerOptions() { } @Test - public void autoEnrollCapabilityImpliesPollerAutoscaling() { - // Auto-enroll implies full autoscaling support (including scale-down), so it also enables the - // pollerAutoscaling flag that PollScaleReportHandle reads as serverSupportsAutoscaling. + public void autoEnrollAndPollerAutoscalingAreIndependent() { + // Auto-enroll drives only the enrollment decision; it does not by itself enable the separate + // pollerAutoscaling (scale-down) capability. The server advertises pollerAutoscaling on its + // own. NamespaceCapabilities caps = capabilities(true, false); assertTrue(caps.isPollerAutoscalingAutoEnroll()); - assertTrue(caps.isPollerAutoscaling()); + assertFalse(caps.isPollerAutoscaling()); } @Test From bd145cc5af821a22f82b97f5747e34ea91ed8752 Mon Sep 17 00:00:00 2001 From: Veeral Patel Date: Tue, 21 Jul 2026 14:14:22 -0700 Subject: [PATCH 4/4] Fix awkward comment wrapping in WorkerPollerAutoEnrollEligibilityTest --- .../worker/WorkerPollerAutoEnrollEligibilityTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java index a72f4bd746..46ae4d37de 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java @@ -126,9 +126,8 @@ public void explicitMaxConcurrentPollersOnOneTypeLeavesOthersEligible() { @Test public void defaultInstanceMakesEveryPollerTypeEligible() { - // getDefaultInstance() carries the numeric default poller count (5), but the user never called - // a - // setter, so all types remain eligible. + // getDefaultInstance() carries the numeric default poller count (5) but no setter was called, + // so all types remain eligible. Worker worker = buildWorker(WorkerOptions.getDefaultInstance()); assertTrue(workflowEligible(worker)); assertTrue(activityEligible(worker));