Fix worker options default application overriding poller behavior

Author: Sushisource

temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java

@@ -646,13 +646,17 @@ public WorkerOptions validateAndBuildWithDefaults() {
           workerTuner,
           maxTaskQueueActivitiesPerSecond,
           maxConcurrentWorkflowTaskPollers == 0
-              ? DEFAULT_MAX_CONCURRENT_WORKFLOW_TASK_POLLERS
+              ? (workflowTaskPollersBehavior != null
+                  ? 0
+                  : DEFAULT_MAX_CONCURRENT_WORKFLOW_TASK_POLLERS)
               : maxConcurrentWorkflowTaskPollers,
           maxConcurrentActivityTaskPollers == 0
-              ? DEFAULT_MAX_CONCURRENT_ACTIVITY_TASK_POLLERS
+              ? (activityTaskPollersBehavior != null
+                  ? 0
+                  : DEFAULT_MAX_CONCURRENT_ACTIVITY_TASK_POLLERS)
               : maxConcurrentActivityTaskPollers,
           maxConcurrentNexusTaskPollers == 0
-              ? DEFAULT_MAX_CONCURRENT_NEXUS_TASK_POLLERS
+              ? (nexusTaskPollersBehavior != null ? 0 : DEFAULT_MAX_CONCURRENT_NEXUS_TASK_POLLERS)
               : maxConcurrentNexusTaskPollers,
           localActivityWorkerOnly,
           defaultDeadlockDetectionTimeout == 0

temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java

@@ -144,6 +144,35 @@ public void throwsIfResourceControllerIsNotSame() {
                 nexusSlotSupplier));
   }
 
+  @Test
+  public void validateAndBuildWithDefaultsDoesNotSetPollerDefaultsWhenBehaviorIsSet() {
+    WorkerOptions options =
+        WorkerOptions.newBuilder()
+            .setWorkflowTaskPollersBehavior(new PollerBehaviorAutoscaling())
+            .setActivityTaskPollersBehavior(new PollerBehaviorAutoscaling())
+            .setNexusTaskPollersBehavior(new PollerBehaviorAutoscaling())
+            .validateAndBuildWithDefaults();
+
+    assertEquals(0, options.getMaxConcurrentWorkflowTaskPollers());
+    assertEquals(0, options.getMaxConcurrentActivityTaskPollers());
+    assertEquals(0, options.getMaxConcurrentNexusTaskPollers());
+    assertNotNull(options.getWorkflowTaskPollersBehavior());
+    assertNotNull(options.getActivityTaskPollersBehavior());
+    assertNotNull(options.getNexusTaskPollersBehavior());
+  }
+
+  @Test
+  public void validateAndBuildWithDefaultsIsIdempotentWithPollerBehavior() {
+    WorkerOptions first =
+        WorkerOptions.newBuilder()
+            .setWorkflowTaskPollersBehavior(new PollerBehaviorAutoscaling())
+            .validateAndBuildWithDefaults();
+    WorkerOptions second = WorkerOptions.newBuilder(first).validateAndBuildWithDefaults();
+
+    assertEquals(0, second.getMaxConcurrentWorkflowTaskPollers());
+    assertNotNull(second.getWorkflowTaskPollersBehavior());
+  }
+
   @Test
   public void verifyMaxTaskQueuePerSecondsDisablesEagerExecution() {
     // Verify that by default eager execution is enabled