From 7ff9ebdf19ffdbd969a09a61b414ebb5cf013c9b Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Wed, 12 Aug 2026 14:41:25 -0700 Subject: [PATCH 1/3] Add test verifying worker command polls omit versioning metadata Add a dedicated test that starts a worker with deployment options and verifies via gRPC interceptor that normal nexus polls carry deploymentOptions while worker-command polls do not. This strengthens the coverage from #2987 where the existing test used an unversioned worker, making the assertions pass trivially. Co-Authored-By: Claude Opus 4.6 --- .../WorkerCommandNexusPollVersioningTest.java | 145 ++++++++++++++++++ 1 file changed, 145 insertions(+) create mode 100644 temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java new file mode 100644 index 000000000..c75c98a33 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java @@ -0,0 +1,145 @@ +package io.temporal.worker; + +import static io.temporal.testUtils.Eventually.assertEventually; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assume.assumeTrue; + +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ForwardingClientCall; +import io.grpc.MethodDescriptor; +import io.temporal.api.enums.v1.TaskQueueKind; +import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest; +import io.temporal.api.workflowservice.v1.PollNexusTaskQueueRequest; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.WorkerDeploymentVersion; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; + +/** + * Verifies that nexus poll requests for worker-command task queues do not carry versioning + * metadata, even when the worker is configured with deployment options. + */ +public class WorkerCommandNexusPollVersioningTest { + + private final List workerCommandPollRequests = + new CopyOnWriteArrayList<>(); + private final List normalNexusPollRequests = + new CopyOnWriteArrayList<>(); + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setTestTimeoutSeconds(15) + .setWorkflowServiceStubsOptions( + WorkflowServiceStubsOptions.newBuilder() + .addGrpcClientInterceptor( + new NexusPollRecordingInterceptor( + workerCommandPollRequests, normalNexusPollRequests)) + .build()) + .setWorkflowClientOptions( + WorkflowClientOptions.newBuilder() + .setWorkerHeartbeatInterval(Duration.ofSeconds(1)) + .build()) + .setWorkerOptions( + WorkerOptions.newBuilder() + .setDeploymentOptions( + WorkerDeploymentOptions.newBuilder() + .setVersion( + new WorkerDeploymentVersion("test-deployment", "test-build-id")) + .build()) + .build()) + .setDoNotStart(true) + .build(); + + @Before + public void checkServerSupportsWorkerCommands() { + assumeTrue( + "Requires real server with worker command support", SDKTestWorkflowRule.useExternalService); + assumeTrue( + "Server does not support worker commands", + testWorkflowRule + .getWorkflowClient() + .getWorkflowServiceStubs() + .blockingStub() + .describeNamespace( + DescribeNamespaceRequest.newBuilder() + .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) + .build()) + .getNamespaceInfo() + .getCapabilities() + .getWorkerCommands()); + } + + @Test + @SuppressWarnings("deprecation") + public void workerCommandPollsOmitVersioningMetadata() { + testWorkflowRule.getTestEnvironment().start(); + + // Wait until we've captured at least one poll of each kind. + assertEventually( + Duration.ofSeconds(10), + () -> { + assertFalse("Expected at least one normal Nexus poll", normalNexusPollRequests.isEmpty()); + assertFalse( + "Expected at least one worker command Nexus poll", + workerCommandPollRequests.isEmpty()); + }); + + // Normal nexus polls must carry deployment options. + for (PollNexusTaskQueueRequest request : normalNexusPollRequests) { + assertTrue( + "Normal nexus poll should have deployment options", request.hasDeploymentOptions()); + } + + // Worker command polls must NOT carry any versioning metadata. + for (PollNexusTaskQueueRequest request : workerCommandPollRequests) { + assertFalse( + "Worker command poll should not have deployment options", request.hasDeploymentOptions()); + assertFalse( + "Worker command poll should not have worker version capabilities", + request.hasWorkerVersionCapabilities()); + } + } + + private static class NexusPollRecordingInterceptor implements ClientInterceptor { + private final List workerCommandPollRequests; + private final List normalNexusPollRequests; + + private NexusPollRecordingInterceptor( + List workerCommandPollRequests, + List normalNexusPollRequests) { + this.workerCommandPollRequests = workerCommandPollRequests; + this.normalNexusPollRequests = normalNexusPollRequests; + } + + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void sendMessage(ReqT message) { + if (message instanceof PollNexusTaskQueueRequest) { + PollNexusTaskQueueRequest request = (PollNexusTaskQueueRequest) message; + if (request.getTaskQueue().getKind() == TaskQueueKind.TASK_QUEUE_KIND_WORKER_COMMANDS) { + workerCommandPollRequests.add(request); + } else if (request.getTaskQueue().getKind() == TaskQueueKind.TASK_QUEUE_KIND_NORMAL) { + normalNexusPollRequests.add(request); + } + } + super.sendMessage(message); + } + }; + } + } +} From b75552606a0c60a9afc97834675c14a7c11c9ed0 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Wed, 12 Aug 2026 17:58:31 -0700 Subject: [PATCH 2/3] Fix: register nexus service and bump timeout - Register EchoNexusServiceImpl so the normal nexus poller starts (without it, NexusTaskHandlerImpl.start() returns false and no normal nexus polls are issued, making the positive assertion fail) - Bump test timeout from 15s to 30s for headroom Co-Authored-By: Claude Opus 4.6 --- .../temporal/worker/WorkerCommandNexusPollVersioningTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java index c75c98a33..413e22125 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java @@ -18,6 +18,7 @@ import io.temporal.common.WorkerDeploymentVersion; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.shared.EchoNexusServiceImpl; import java.time.Duration; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -39,7 +40,7 @@ public class WorkerCommandNexusPollVersioningTest { @Rule public SDKTestWorkflowRule testWorkflowRule = SDKTestWorkflowRule.newBuilder() - .setTestTimeoutSeconds(15) + .setTestTimeoutSeconds(30) .setWorkflowServiceStubsOptions( WorkflowServiceStubsOptions.newBuilder() .addGrpcClientInterceptor( @@ -58,6 +59,7 @@ public class WorkerCommandNexusPollVersioningTest { new WorkerDeploymentVersion("test-deployment", "test-build-id")) .build()) .build()) + .setNexusServiceImplementation(new EchoNexusServiceImpl()) .setDoNotStart(true) .build(); From 42198e3434cbccd5c5c048340e048724ddc8e628 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 13 Aug 2026 10:34:02 -0700 Subject: [PATCH 3/3] Fix existing test instead of adding a separate one Configure deployment options (without useVersioning) on the existing ActivityCancellationTokenIntegrationTest worker, register a nexus service so the normal poller starts, and assert both sides: normal polls carry deploymentOptions, worker-command polls do not. Co-Authored-By: Claude Opus 4.6 --- .../WorkerCommandNexusPollVersioningTest.java | 147 ------------------ ...ivityCancellationTokenIntegrationTest.java | 53 ++++++- 2 files changed, 46 insertions(+), 154 deletions(-) delete mode 100644 temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java deleted file mode 100644 index 413e22125..000000000 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerCommandNexusPollVersioningTest.java +++ /dev/null @@ -1,147 +0,0 @@ -package io.temporal.worker; - -import static io.temporal.testUtils.Eventually.assertEventually; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; -import static org.junit.Assume.assumeTrue; - -import io.grpc.CallOptions; -import io.grpc.Channel; -import io.grpc.ClientCall; -import io.grpc.ClientInterceptor; -import io.grpc.ForwardingClientCall; -import io.grpc.MethodDescriptor; -import io.temporal.api.enums.v1.TaskQueueKind; -import io.temporal.api.workflowservice.v1.DescribeNamespaceRequest; -import io.temporal.api.workflowservice.v1.PollNexusTaskQueueRequest; -import io.temporal.client.WorkflowClientOptions; -import io.temporal.common.WorkerDeploymentVersion; -import io.temporal.serviceclient.WorkflowServiceStubsOptions; -import io.temporal.testing.internal.SDKTestWorkflowRule; -import io.temporal.workflow.shared.EchoNexusServiceImpl; -import java.time.Duration; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; - -/** - * Verifies that nexus poll requests for worker-command task queues do not carry versioning - * metadata, even when the worker is configured with deployment options. - */ -public class WorkerCommandNexusPollVersioningTest { - - private final List workerCommandPollRequests = - new CopyOnWriteArrayList<>(); - private final List normalNexusPollRequests = - new CopyOnWriteArrayList<>(); - - @Rule - public SDKTestWorkflowRule testWorkflowRule = - SDKTestWorkflowRule.newBuilder() - .setTestTimeoutSeconds(30) - .setWorkflowServiceStubsOptions( - WorkflowServiceStubsOptions.newBuilder() - .addGrpcClientInterceptor( - new NexusPollRecordingInterceptor( - workerCommandPollRequests, normalNexusPollRequests)) - .build()) - .setWorkflowClientOptions( - WorkflowClientOptions.newBuilder() - .setWorkerHeartbeatInterval(Duration.ofSeconds(1)) - .build()) - .setWorkerOptions( - WorkerOptions.newBuilder() - .setDeploymentOptions( - WorkerDeploymentOptions.newBuilder() - .setVersion( - new WorkerDeploymentVersion("test-deployment", "test-build-id")) - .build()) - .build()) - .setNexusServiceImplementation(new EchoNexusServiceImpl()) - .setDoNotStart(true) - .build(); - - @Before - public void checkServerSupportsWorkerCommands() { - assumeTrue( - "Requires real server with worker command support", SDKTestWorkflowRule.useExternalService); - assumeTrue( - "Server does not support worker commands", - testWorkflowRule - .getWorkflowClient() - .getWorkflowServiceStubs() - .blockingStub() - .describeNamespace( - DescribeNamespaceRequest.newBuilder() - .setNamespace(testWorkflowRule.getWorkflowClient().getOptions().getNamespace()) - .build()) - .getNamespaceInfo() - .getCapabilities() - .getWorkerCommands()); - } - - @Test - @SuppressWarnings("deprecation") - public void workerCommandPollsOmitVersioningMetadata() { - testWorkflowRule.getTestEnvironment().start(); - - // Wait until we've captured at least one poll of each kind. - assertEventually( - Duration.ofSeconds(10), - () -> { - assertFalse("Expected at least one normal Nexus poll", normalNexusPollRequests.isEmpty()); - assertFalse( - "Expected at least one worker command Nexus poll", - workerCommandPollRequests.isEmpty()); - }); - - // Normal nexus polls must carry deployment options. - for (PollNexusTaskQueueRequest request : normalNexusPollRequests) { - assertTrue( - "Normal nexus poll should have deployment options", request.hasDeploymentOptions()); - } - - // Worker command polls must NOT carry any versioning metadata. - for (PollNexusTaskQueueRequest request : workerCommandPollRequests) { - assertFalse( - "Worker command poll should not have deployment options", request.hasDeploymentOptions()); - assertFalse( - "Worker command poll should not have worker version capabilities", - request.hasWorkerVersionCapabilities()); - } - } - - private static class NexusPollRecordingInterceptor implements ClientInterceptor { - private final List workerCommandPollRequests; - private final List normalNexusPollRequests; - - private NexusPollRecordingInterceptor( - List workerCommandPollRequests, - List normalNexusPollRequests) { - this.workerCommandPollRequests = workerCommandPollRequests; - this.normalNexusPollRequests = normalNexusPollRequests; - } - - @Override - public ClientCall interceptCall( - MethodDescriptor method, CallOptions callOptions, Channel next) { - return new ForwardingClientCall.SimpleForwardingClientCall( - next.newCall(method, callOptions)) { - @Override - public void sendMessage(ReqT message) { - if (message instanceof PollNexusTaskQueueRequest) { - PollNexusTaskQueueRequest request = (PollNexusTaskQueueRequest) message; - if (request.getTaskQueue().getKind() == TaskQueueKind.TASK_QUEUE_KIND_WORKER_COMMANDS) { - workerCommandPollRequests.add(request); - } else if (request.getTaskQueue().getKind() == TaskQueueKind.TASK_QUEUE_KIND_NORMAL) { - normalNexusPollRequests.add(request); - } - } - super.sendMessage(message); - } - }; - } - } -} diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/cancellation/ActivityCancellationTokenIntegrationTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/cancellation/ActivityCancellationTokenIntegrationTest.java index d704e186b..26dfc88da 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/cancellation/ActivityCancellationTokenIntegrationTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/cancellation/ActivityCancellationTokenIntegrationTest.java @@ -2,6 +2,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeTrue; import io.grpc.CallOptions; @@ -21,10 +22,13 @@ import io.temporal.api.workflowservice.v1.PollNexusTaskQueueRequest; import io.temporal.client.ActivityCanceledException; import io.temporal.client.WorkflowClientOptions; +import io.temporal.common.WorkerDeploymentVersion; import io.temporal.failure.ActivityFailure; import io.temporal.failure.CanceledFailure; import io.temporal.serviceclient.WorkflowServiceStubsOptions; import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.worker.WorkerDeploymentOptions; +import io.temporal.worker.WorkerOptions; import io.temporal.workflow.Async; import io.temporal.workflow.CancellationScope; import io.temporal.workflow.Promise; @@ -32,6 +36,7 @@ import io.temporal.workflow.Workflow; import io.temporal.workflow.WorkflowInterface; import io.temporal.workflow.WorkflowMethod; +import io.temporal.workflow.shared.EchoNexusServiceImpl; import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -47,6 +52,8 @@ public class ActivityCancellationTokenIntegrationTest { private final List workerCommandPollRequests = new CopyOnWriteArrayList<>(); + private final List normalNexusPollRequests = + new CopyOnWriteArrayList<>(); @Rule public SDKTestWorkflowRule testWorkflowRule = @@ -55,7 +62,21 @@ public class ActivityCancellationTokenIntegrationTest { .setWorkflowServiceStubsOptions( WorkflowServiceStubsOptions.newBuilder() .addGrpcClientInterceptor( - new WorkerCommandPollRecordingInterceptor(workerCommandPollRequests)) + new NexusPollRecordingInterceptor( + workerCommandPollRequests, normalNexusPollRequests)) + .build()) + // Configure deployment options without useVersioning(true). This causes the SDK to + // send deploymentOptions on poll requests (with UNVERSIONED mode) without requiring + // server-side deployment setup. This is sufficient to verify that worker-command polls + // omit deploymentOptions while normal polls include them — the field presence is the + // same regardless of versioning mode. + .setWorkerOptions( + WorkerOptions.newBuilder() + .setDeploymentOptions( + WorkerDeploymentOptions.newBuilder() + .setVersion( + new WorkerDeploymentVersion("test-deployment", "test-build-id")) + .build()) .build()) .setWorkflowClientOptions( WorkflowClientOptions.newBuilder() @@ -63,6 +84,7 @@ public class ActivityCancellationTokenIntegrationTest { .build()) .setWorkflowTypes(TestCancellationWorkflowImpl.class) .setActivityImplementations(new NonHeartbeatingActivityImpl()) + .setNexusServiceImplementation(new EchoNexusServiceImpl()) .build(); @Before @@ -95,19 +117,34 @@ public void activityObservesCancellationWithoutHeartbeat() { assertEquals("cancelled", workflow.execute(testWorkflowRule.getTaskQueue())); - assertFalse("Expected a worker command Nexus poll", workerCommandPollRequests.isEmpty()); + // Normal nexus polls must carry deployment options (positive control). + assertFalse("Expected at least one normal Nexus poll", normalNexusPollRequests.isEmpty()); + for (PollNexusTaskQueueRequest request : normalNexusPollRequests) { + assertTrue( + "Normal nexus poll should have deployment options", request.hasDeploymentOptions()); + } + + // Worker command polls must NOT carry any versioning metadata. + assertFalse( + "Expected at least one worker command Nexus poll", workerCommandPollRequests.isEmpty()); for (PollNexusTaskQueueRequest request : workerCommandPollRequests) { - assertFalse(request.hasDeploymentOptions()); - assertFalse(request.hasWorkerVersionCapabilities()); + assertFalse( + "Worker command poll should not have deployment options", request.hasDeploymentOptions()); + assertFalse( + "Worker command poll should not have worker version capabilities", + request.hasWorkerVersionCapabilities()); } } - private static class WorkerCommandPollRecordingInterceptor implements ClientInterceptor { + private static class NexusPollRecordingInterceptor implements ClientInterceptor { private final List workerCommandPollRequests; + private final List normalNexusPollRequests; - private WorkerCommandPollRecordingInterceptor( - List workerCommandPollRequests) { + private NexusPollRecordingInterceptor( + List workerCommandPollRequests, + List normalNexusPollRequests) { this.workerCommandPollRequests = workerCommandPollRequests; + this.normalNexusPollRequests = normalNexusPollRequests; } @Override @@ -121,6 +158,8 @@ public void sendMessage(ReqT message) { PollNexusTaskQueueRequest request = (PollNexusTaskQueueRequest) message; if (request.getTaskQueue().getKind() == TaskQueueKind.TASK_QUEUE_KIND_WORKER_COMMANDS) { workerCommandPollRequests.add(request); + } else if (request.getTaskQueue().getKind() == TaskQueueKind.TASK_QUEUE_KIND_NORMAL) { + normalNexusPollRequests.add(request); } } super.sendMessage(message);