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 6c86fc447..ff528d46b 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 ed4ac3935..4bddd45d9 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,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(); } 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 1fd9cf914..33416a807 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 1765c5d1c..1245907dd 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 0de3ffaf6..b86dfea6d 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 5ba761386..93a26def2 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,15 @@ private static final class TaskSnapshot { Map tags = new ImmutableMap.Builder(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 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 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 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 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 +1022,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 4e351cb74..99a596514 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; } /** @@ -229,9 +239,14 @@ 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; + this.workflowTaskPollersConfigured = true; return this; } @@ -241,10 +256,15 @@ 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) { this.maxConcurrentNexusTaskPollers = maxConcurrentNexusTaskPollers; + this.nexusTaskPollersConfigured = true; return this; } @@ -268,9 +288,14 @@ 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; + this.activityTaskPollersConfigured = true; return this; } @@ -505,21 +530,43 @@ 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; + this.workflowTaskPollersConfigured = true; 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; + this.activityTaskPollersConfigured = true; 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; + this.nexusTaskPollersConfigured = true; return this; } @@ -589,7 +636,10 @@ public WorkerOptions build() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, - preferredVersionProvider); + preferredVersionProvider, + workflowTaskPollersConfigured, + activityTaskPollersConfigured, + nexusTaskPollersConfigured); } public WorkerOptions validateAndBuildWithDefaults() { @@ -723,7 +773,10 @@ public WorkerOptions validateAndBuildWithDefaults() { activityTaskPollersBehavior, nexusTaskPollersBehavior, allowActivityHeartbeatDuringShutdown, - preferredVersionProvider); + preferredVersionProvider, + workflowTaskPollersConfigured, + activityTaskPollersConfigured, + nexusTaskPollersConfigured); } } @@ -757,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, @@ -788,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; @@ -819,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/internal/worker/PollerAutoscalingAutoEnrollTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java new file mode 100644 index 000000000..5d0df9cc7 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/PollerAutoscalingAutoEnrollTest.java @@ -0,0 +1,114 @@ +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 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()); + assertFalse(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 000000000..46ae4d37d --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerPollerAutoEnrollEligibilityTest.java @@ -0,0 +1,185 @@ +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 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 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 { + + 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 defaultInstanceMakesEveryPollerTypeEligible() { + // 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)); + 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 = + 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-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 000000000..00ea0d69b --- /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); + } + } +} diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index d2fc34ab8..852ee3b33 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit d2fc34ab844603f50e41365f46c7fb82bdedffe6 +Subproject commit 852ee3b339f9f1efe3503fe96f0351d361396daa