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..158e46e25 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 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..32191b34a 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 4e351cb74..d181a75ea 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 000000000..556f722fa --- /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 000000000..949491d10 --- /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 d2fc34ab8..852ee3b33 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit d2fc34ab844603f50e41365f46c7fb82bdedffe6 +Subproject commit 852ee3b339f9f1efe3503fe96f0351d361396daa