Skip to content

Commit 6d09a34

Browse files
veeral-patelveeral-patel
authored
Auto-enroll pollers into autoscaling on PollerAutoscalingAutoEnroll (Java SDK) (#2953)
* 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 MaxConcurrent<Type>TaskPollers nor a <Type>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). * Address review: track poller-setter intent for auto-enroll eligibility - Determine auto-enroll eligibility from whether the user actually called a poller setter (MaxConcurrent<Type>TaskPollers or <Type>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. * 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. * Fix awkward comment wrapping in WorkerPollerAutoEnrollEligibilityTest
1 parent 8b57794 commit 6d09a34

11 files changed

Lines changed: 647 additions & 17 deletions

File tree

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ final class ActivityWorker implements SuspendableWorker {
4343
private final String taskQueue;
4444
private final SingleWorkerOptions options;
4545
private final double taskQueueActivitiesPerSecond;
46-
private final PollerOptions pollerOptions;
46+
private PollerOptions pollerOptions;
4747
private final Scope workerMetricsScope;
4848
private final GrpcRetryer grpcRetryer;
4949
private final GrpcRetryer.GrpcRetryerOptions replyGrpcRetryerOptions;
@@ -83,6 +83,11 @@ public ActivityWorker(
8383
@Override
8484
public boolean start() {
8585
if (handler.isAnyTypeSupported()) {
86+
// Auto-enroll into poller autoscaling if the namespace advertises the capability and this
87+
// poller type was left at its default. Resolved here (after namespace capabilities are known)
88+
// so the poller built below reflects the effective behavior.
89+
this.pollerOptions =
90+
PollerOptions.maybeEnrollInPollerAutoscaling(pollerOptions, namespaceCapabilities);
8691
this.pollTaskExecutor =
8792
new PollTaskExecutor<>(
8893
namespace,

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,15 @@
1010
*/
1111
public final class NamespaceCapabilities {
1212
private final AtomicBoolean pollerAutoscaling = new AtomicBoolean(false);
13+
private final AtomicBoolean pollerAutoscalingAutoEnroll = new AtomicBoolean(false);
1314
private final AtomicBoolean gracefulPollShutdown = new AtomicBoolean(false);
1415
private final AtomicBoolean workerHeartbeats = new AtomicBoolean(false);
1516
private final AtomicBoolean workerCommands = new AtomicBoolean(false);
1617

1718
public void setFromCapabilities(Capabilities capabilities) {
19+
if (capabilities.getPollerAutoscalingAutoEnroll()) {
20+
pollerAutoscalingAutoEnroll.set(true);
21+
}
1822
if (capabilities.getPollerAutoscaling()) {
1923
pollerAutoscaling.set(true);
2024
}
@@ -33,6 +37,10 @@ public boolean isPollerAutoscaling() {
3337
return pollerAutoscaling.get();
3438
}
3539

40+
public boolean isPollerAutoscalingAutoEnroll() {
41+
return pollerAutoscalingAutoEnroll.get();
42+
}
43+
3644
public boolean isGracefulPollShutdown() {
3745
return gracefulPollShutdown.get();
3846
}

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ final class NexusWorker implements SuspendableWorker {
4646
private final String namespace;
4747
private final String taskQueue;
4848
private final SingleWorkerOptions options;
49-
private final PollerOptions pollerOptions;
49+
private PollerOptions pollerOptions;
5050
private final Scope workerMetricsScope;
5151
private final DataConverter dataConverter;
5252
private final GrpcRetryer grpcRetryer;
@@ -114,6 +114,11 @@ public NexusWorker(
114114
@Override
115115
public boolean start() {
116116
if (handler.start()) {
117+
// Auto-enroll into poller autoscaling if the namespace advertises the capability and this
118+
// poller type was left at its default. Resolved here (after namespace capabilities are known)
119+
// so the poller built below reflects the effective behavior.
120+
this.pollerOptions =
121+
PollerOptions.maybeEnrollInPollerAutoscaling(pollerOptions, namespaceCapabilities);
117122
this.pollTaskExecutor =
118123
new PollTaskExecutor<>(
119124
namespace,

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

Lines changed: 44 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import io.grpc.Status;
44
import io.grpc.StatusRuntimeException;
55
import io.temporal.worker.tuning.PollerBehavior;
6+
import io.temporal.worker.tuning.PollerBehaviorAutoscaling;
67
import java.time.Duration;
78
import java.util.concurrent.ExecutorService;
89
import org.slf4j.Logger;
@@ -26,6 +27,27 @@ public static PollerOptions getDefaultInstance() {
2627
return DEFAULT_INSTANCE;
2728
}
2829

30+
/**
31+
* If the given options are eligible for poller-autoscaling auto-enrollment (the user left this
32+
* poller type at its default) and the namespace advertises the auto-enroll capability, returns a
33+
* copy of the options with a default {@link PollerBehaviorAutoscaling} behavior. Otherwise
34+
* returns the options unchanged.
35+
*
36+
* <p>Must only be called before the worker's poller is created (i.e. at worker start), so the
37+
* resolved behavior is picked up when the poller is built.
38+
*/
39+
public static PollerOptions maybeEnrollInPollerAutoscaling(
40+
PollerOptions options, NamespaceCapabilities namespaceCapabilities) {
41+
if (options.isAutoscalingAutoEnrollEligible()
42+
&& namespaceCapabilities.isPollerAutoscalingAutoEnroll()
43+
&& !(options.getPollerBehavior() instanceof PollerBehaviorAutoscaling)) {
44+
return PollerOptions.newBuilder(options)
45+
.setPollerBehavior(new PollerBehaviorAutoscaling())
46+
.build();
47+
}
48+
return options;
49+
}
50+
2951
private static final PollerOptions DEFAULT_INSTANCE;
3052

3153
static {
@@ -46,6 +68,7 @@ public static final class Builder {
4668
private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
4769
private boolean usingVirtualThreads;
4870
private ExecutorService pollerTaskExecutorOverride;
71+
private boolean autoscalingAutoEnrollEligible;
4972

5073
private Builder() {}
5174

@@ -65,6 +88,7 @@ private Builder(PollerOptions options) {
6588
this.uncaughtExceptionHandler = options.getUncaughtExceptionHandler();
6689
this.usingVirtualThreads = options.isUsingVirtualThreads();
6790
this.pollerTaskExecutorOverride = options.getPollerTaskExecutorOverride();
91+
this.autoscalingAutoEnrollEligible = options.isAutoscalingAutoEnrollEligible();
6892
}
6993

7094
/** Defines interval for measuring poll rate. Larger the interval more spiky can be the load. */
@@ -152,6 +176,16 @@ public Builder setPollerTaskExecutorOverride(ExecutorService overrideTaskExecuto
152176
return this;
153177
}
154178

179+
/**
180+
* Marks whether this poller type was left at its default (the user set neither a fixed poller
181+
* count nor a poller behavior) and is therefore eligible for poller-autoscaling auto-enrollment
182+
* when the namespace advertises the capability.
183+
*/
184+
public Builder setAutoscalingAutoEnrollEligible(boolean autoscalingAutoEnrollEligible) {
185+
this.autoscalingAutoEnrollEligible = autoscalingAutoEnrollEligible;
186+
return this;
187+
}
188+
155189
public PollerOptions build() {
156190
if (uncaughtExceptionHandler == null) {
157191
uncaughtExceptionHandler =
@@ -180,7 +214,8 @@ public PollerOptions build() {
180214
uncaughtExceptionHandler,
181215
pollThreadNamePrefix,
182216
usingVirtualThreads,
183-
pollerTaskExecutorOverride);
217+
pollerTaskExecutorOverride,
218+
autoscalingAutoEnrollEligible);
184219
}
185220
}
186221

@@ -198,6 +233,7 @@ public PollerOptions build() {
198233
private final boolean usingVirtualThreads;
199234
private final ExecutorService pollerTaskExecutorOverride;
200235
private final PollerBehavior pollerBehavior;
236+
private final boolean autoscalingAutoEnrollEligible;
201237

202238
private PollerOptions(
203239
int maximumPollRateIntervalMilliseconds,
@@ -211,7 +247,8 @@ private PollerOptions(
211247
Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
212248
String pollThreadNamePrefix,
213249
boolean usingVirtualThreads,
214-
ExecutorService pollerTaskExecutorOverride) {
250+
ExecutorService pollerTaskExecutorOverride,
251+
boolean autoscalingAutoEnrollEligible) {
215252
this.maximumPollRateIntervalMilliseconds = maximumPollRateIntervalMilliseconds;
216253
this.maximumPollRatePerSecond = maximumPollRatePerSecond;
217254
this.backoffCoefficient = backoffCoefficient;
@@ -224,6 +261,7 @@ private PollerOptions(
224261
this.pollThreadNamePrefix = pollThreadNamePrefix;
225262
this.usingVirtualThreads = usingVirtualThreads;
226263
this.pollerTaskExecutorOverride = pollerTaskExecutorOverride;
264+
this.autoscalingAutoEnrollEligible = autoscalingAutoEnrollEligible;
227265
}
228266

229267
public int getMaximumPollRateIntervalMilliseconds() {
@@ -274,6 +312,10 @@ public ExecutorService getPollerTaskExecutorOverride() {
274312
return pollerTaskExecutorOverride;
275313
}
276314

315+
public boolean isAutoscalingAutoEnrollEligible() {
316+
return autoscalingAutoEnrollEligible;
317+
}
318+
277319
@Override
278320
public String toString() {
279321
return "PollerOptions{"

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ final class WorkflowWorker implements SuspendableWorker {
4848
private final WorkflowExecutorCache cache;
4949
private final WorkflowTaskHandler handler;
5050
private final String stickyTaskQueueName;
51-
private final PollerOptions pollerOptions;
51+
private PollerOptions pollerOptions;
5252
private final Scope workerMetricsScope;
5353
private final GrpcRetryer grpcRetryer;
5454
private final EagerActivityDispatcher eagerActivityDispatcher;
@@ -99,6 +99,11 @@ public WorkflowWorker(
9999
@Override
100100
public boolean start() {
101101
if (handler.isAnyTypeSupported()) {
102+
// Auto-enroll into poller autoscaling if the namespace advertises the capability and this
103+
// poller type was left at its default. Resolved here (after namespace capabilities are known)
104+
// so the poller built below reflects the effective behavior.
105+
this.pollerOptions =
106+
PollerOptions.maybeEnrollInPollerAutoscaling(pollerOptions, namespaceCapabilities);
102107
pollTaskExecutor =
103108
new PollTaskExecutor<>(
104109
namespace,

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

Lines changed: 24 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,15 @@ private static final class TaskSnapshot {
132132
Map<String, String> tags =
133133
new ImmutableMap.Builder<String, String>(1).put(MetricsTag.TASK_QUEUE, taskQueue).build();
134134
Scope taggedScope = metricsScope.tagged(tags);
135+
136+
// Poller types the user left at their default are auto-enrolled into poller autoscaling at
137+
// start() when the namespace advertises the PollerAutoscalingAutoEnroll capability. Eligibility
138+
// tracks whether the user called a poller setter (recorded on WorkerOptions.Builder), not the
139+
// resolved value, so a defaulted count of 5 is not mistaken for an explicit choice.
140+
boolean workflowTaskAutoEnrollEligible = this.options.isWorkflowTaskPollerAutoEnrollEligible();
141+
boolean activityTaskAutoEnrollEligible = this.options.isActivityTaskPollerAutoEnrollEligible();
142+
boolean nexusTaskAutoEnrollEligible = this.options.isNexusTaskPollerAutoEnrollEligible();
143+
135144
SingleWorkerOptions activityOptions =
136145
toActivityOptions(
137146
factoryOptions,
@@ -140,7 +149,8 @@ private static final class TaskSnapshot {
140149
contextPropagators,
141150
taggedScope,
142151
workerInstanceKey,
143-
workerControlTaskQueue);
152+
workerControlTaskQueue,
153+
activityTaskAutoEnrollEligible);
144154
if (this.options.isLocalActivityWorkerOnly()) {
145155
activityWorker = null;
146156
} else {
@@ -174,7 +184,8 @@ private static final class TaskSnapshot {
174184
contextPropagators,
175185
taggedScope,
176186
workerInstanceKey,
177-
workerControlTaskQueue);
187+
workerControlTaskQueue,
188+
nexusTaskAutoEnrollEligible);
178189
SlotSupplier<NexusSlotInfo> nexusSlotSupplier =
179190
this.options.getWorkerTuner() == null
180191
? new FixedSizeSlotSupplier<>(this.options.getMaxConcurrentNexusExecutionSize())
@@ -194,7 +205,8 @@ private static final class TaskSnapshot {
194205
contextPropagators,
195206
taggedScope,
196207
workerInstanceKey,
197-
workerControlTaskQueue);
208+
workerControlTaskQueue,
209+
workflowTaskAutoEnrollEligible);
198210
SingleWorkerOptions localActivityOptions =
199211
toLocalActivityOptions(
200212
factoryOptions,
@@ -901,7 +913,8 @@ private static SingleWorkerOptions toActivityOptions(
901913
List<ContextPropagator> contextPropagators,
902914
Scope metricsScope,
903915
String workerInstanceKey,
904-
String workerControlTaskQueue) {
916+
String workerControlTaskQueue,
917+
boolean autoEnrollEligible) {
905918
return toSingleWorkerOptions(
906919
factoryOptions,
907920
options,
@@ -920,6 +933,7 @@ private static SingleWorkerOptions toActivityOptions(
920933
: new PollerBehaviorSimpleMaximum(
921934
options.getMaxConcurrentActivityTaskPollers()))
922935
.setUsingVirtualThreads(options.isUsingVirtualThreadsOnActivityWorker())
936+
.setAutoscalingAutoEnrollEligible(autoEnrollEligible)
923937
.build())
924938
.setMetricsScope(metricsScope)
925939
.build();
@@ -932,7 +946,8 @@ private static SingleWorkerOptions toNexusOptions(
932946
List<ContextPropagator> contextPropagators,
933947
Scope metricsScope,
934948
String workerInstanceKey,
935-
String workerControlTaskQueue) {
949+
String workerControlTaskQueue,
950+
boolean autoEnrollEligible) {
936951
return toSingleWorkerOptions(
937952
factoryOptions,
938953
options,
@@ -948,6 +963,7 @@ private static SingleWorkerOptions toNexusOptions(
948963
: new PollerBehaviorSimpleMaximum(
949964
options.getMaxConcurrentNexusTaskPollers()))
950965
.setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
966+
.setAutoscalingAutoEnrollEligible(autoEnrollEligible)
951967
.build())
952968
.setMetricsScope(metricsScope)
953969
.setUsingVirtualThreads(options.isUsingVirtualThreadsOnNexusWorker())
@@ -962,7 +978,8 @@ private static SingleWorkerOptions toWorkflowWorkerOptions(
962978
List<ContextPropagator> contextPropagators,
963979
Scope metricsScope,
964980
String workerInstanceKey,
965-
String workerControlTaskQueue) {
981+
String workerControlTaskQueue,
982+
boolean autoEnrollEligible) {
966983
Map<String, String> tags =
967984
new ImmutableMap.Builder<String, String>(1).put(MetricsTag.TASK_QUEUE, taskQueue).build();
968985

@@ -1005,6 +1022,7 @@ private static SingleWorkerOptions toWorkflowWorkerOptions(
10051022
? pollerBehavior
10061023
: new PollerBehaviorSimpleMaximum(maxConcurrentWorkflowTaskPollers))
10071024
.setUsingVirtualThreads(options.isUsingVirtualThreadsOnWorkflowWorker())
1025+
.setAutoscalingAutoEnrollEligible(autoEnrollEligible)
10081026
.build())
10091027
.setStickyQueueScheduleToStartTimeout(stickyQueueScheduleToStartTimeout)
10101028
.setStickyTaskQueueDrainTimeout(options.getStickyTaskQueueDrainTimeout())

0 commit comments

Comments
 (0)