From 8fdbb33178c4c73d0966ae6ecd8160f5e36ddeba Mon Sep 17 00:00:00 2001 From: tekkaya <86028633+tekkaya@users.noreply.github.com> Date: Thu, 27 Aug 2026 17:42:54 -0700 Subject: [PATCH 1/2] Support starting multiple activities from a synchronous Nexus operation handler TemporalNexusClient guards its own activity/workflow/update start to at most one per operation invocation, so a synchronous Nexus operation handler that starts two or more activities inline has to bypass it and use a raw ActivityClient obtained from Nexus.getOperationContext() instead. That bypass path got no request links, because RootActivityClientInvoker.startActivity derived link attachment, on-conflict dedup, and completion-callback attachment all from the same NexusOperationMetadata, which only the guarded TemporalNexusClient call ever sets. Every activity start made during the invocation now reuses the inbound Nexus task's request ID and gets its inbound links, regardless of which client object issued it. NexusOperationMetadata keeps its narrow, one-shot scope and remains the only thing that can attach a completion callback, since only the guarded start should complete the Nexus operation. When metadata is present its own request ID still takes precedence, so the guarded start's identity never depends on the ambient value also having been set. Reusing one ambient ID across every start in an invocation is a deliberate, known tradeoff: a handler that starts a fresh run under an activity ID it already used earlier in the same invocation can have that start incorrectly resolve to the stale run instead of creating a new one. Two narrower, redelivery-aware alternatives were tried and dropped -- raw ambient-ID reuse scoped to only the guarded call, then a per-call ID derived from each start's ordinal position -- because both require assuming the handler reissues an identical sequence of calls on every retry, an assumption the SDK has no way to verify. This change instead matches sdk-go's approved fix (temporalnexus/temporal_operation.go, PR #2633) and sdk-python's current behavior (temporalio/nexus/_operation_context.py): every activity start in a Nexus context reuses the ambient request ID unconditionally. sdk-python's own attempt at the narrower, backing-call-only scoping (PR #1722) was closed without merging for the same reason. ActivityOperationLinkingTest (functional, requires a real server) drives a synchronous handler that starts two bypass-path activities and asserts both the forward link (each activity's own ActivityExecutionInfo) and the backward links (both activities' completions landing on the caller's single NexusOperationCompleted event), the same way SignalOperationLinkingTest already does for signals. RootActivityClientInvokerTest covers metadata's request ID taking precedence over the ambient one, the ambient-links-and-request-ID-without-metadata case, the outside-Nexus-context case, and two bypass-path starts in one invocation sharing the ambient request ID. --- .../client/RootActivityClientInvoker.java | 38 +++- .../nexus/InternalNexusOperationContext.java | 16 ++ .../internal/nexus/NexusTaskHandlerImpl.java | 3 + .../client/RootActivityClientInvokerTest.java | 74 ++++++- .../nexus/ActivityOperationLinkingTest.java | 191 ++++++++++++++++++ 5 files changed, 310 insertions(+), 12 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/workflow/nexus/ActivityOperationLinkingTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 225228e6a2..902a847b2c 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -64,14 +64,21 @@ public StartActivityOutput startActivity(StartActivityInput input) { NexusOperationMetadata nexusOperationMetadata = nexusContext == null ? null : nexusContext.getNexusOperationMetadata(); + String requestId; + if (nexusOperationMetadata != null + && !Strings.isNullOrEmpty(nexusOperationMetadata.requestId)) { + requestId = nexusOperationMetadata.requestId; + } else if (nexusContext != null && !Strings.isNullOrEmpty(nexusContext.getRequestId())) { + requestId = nexusContext.getRequestId(); + } else { + requestId = UUID.randomUUID().toString(); + } + StartActivityExecutionRequest.Builder request = StartActivityExecutionRequest.newBuilder() .setNamespace(clientOptions.getNamespace()) .setIdentity(clientOptions.getIdentity()) - .setRequestId( - nexusOperationMetadata == null - ? UUID.randomUUID().toString() - : nexusOperationMetadata.requestId) + .setRequestId(requestId) .setActivityId(options.getId()) .setActivityType(ActivityType.newBuilder().setName(input.getActivityType()).build()) .setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build()) @@ -121,14 +128,25 @@ public StartActivityOutput startActivity(StartActivityInput input) { io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null); request.setHeader(grpcHeader); - if (nexusOperationMetadata != null) { - List protoLinks = nexusContext.getRequestLinks(); + List protoLinks = Collections.emptyList(); + if (nexusContext != null) { + // Propagate the inbound Nexus request ID and links to every activity start on the + // operation-handler thread, including starts through a raw ActivityClient. + // Completion callbacks remain limited to the metadata-backed start because only it + // completes the Nexus operation. + protoLinks = nexusContext.getRequestLinks(); request.addAllLinks(protoLinks); - request.setOnConflictOptions( + io.temporal.api.common.v1.OnConflictOptions.Builder onConflictOptions = io.temporal.api.common.v1.OnConflictOptions.newBuilder() .setAttachRequestId(true) - .setAttachLinks(true) - .setAttachCompletionCallbacks(true)); + .setAttachLinks(true); + if (nexusOperationMetadata != null) { + onConflictOptions.setAttachCompletionCallbacks(true); + } + request.setOnConflictOptions(onConflictOptions); + } + + if (nexusOperationMetadata != null) { // Generate the operation token from the user-supplied activity ID and namespace so the // dual OPERATION_ID + OPERATION_TOKEN headers can be injected before the start RPC fires. try { @@ -168,7 +186,7 @@ public StartActivityOutput startActivity(StartActivityInput input) { throw e; } - if (nexusOperationMetadata != null && response.hasLink()) { + if (nexusContext != null && response.hasLink()) { nexusContext.addResponseLink(response.getLink()); } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java index 97ab4f5ed2..3c5a6b0af8 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java @@ -27,6 +27,12 @@ public class InternalNexusOperationContext { // workflow client can attach them to the outgoing requests it issues (e.g. signal, // signalWithStart) via the request's links field. private List requestLinks = Collections.emptyList(); + // The inbound Nexus task's request ID, captured at the task-handler boundary and available to + // clients executing on the operation-handler thread. RootActivityClientInvoker reuses it for + // redelivery-safe activity-start deduplication. It is deliberately independent of + // nexusOperationMetadata, which is scoped to the single backing start because it carries + // completion-callback semantics. + private String requestId; // Links returned by outbound RPCs the operation handler issues (such as // SignalWorkflowExecutionResponse.link or SignalWithStartWorkflowExecutionResponse.signal_link). // One entry per outbound RPC that returned a link. Drained @@ -106,6 +112,16 @@ public void setRequestLinks(List links) { return Collections.unmodifiableList(requestLinks); } + /** Set the request ID of the inbound Nexus task, ambient for the whole invocation. */ + public void setRequestId(String requestId) { + this.requestId = requestId; + } + + /** The inbound Nexus task's request ID; {@code null} if not set. */ + public String getRequestId() { + return requestId; + } + public void setStartWorkflowResponseLink(Link link) { this.startWorkflowResponseLink = link; } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java index 4d40183c27..5ef945ec79 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java @@ -313,6 +313,9 @@ private StartOperationResponse handleStartOperation( } }); CurrentNexusOperationContext.get().setRequestLinks(inboundCommonLinks); + // Ambient for the whole operation-handler invocation, independent of NexusOperationMetadata. + // see InternalNexusOperationContext.requestId. + CurrentNexusOperationContext.get().setRequestId(task.getRequestId()); HandlerInputContent.Builder input = HandlerInputContent.newBuilder().setDataStream(task.getPayload().toByteString().newInput()); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java index 16ba8b7f36..28bfe74593 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java @@ -2,6 +2,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -22,6 +23,7 @@ import java.time.Duration; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import org.junit.After; import org.junit.Assert; @@ -106,6 +108,21 @@ public void nexusMetadataAddsCallbackLinksAndRequestId() { Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks()); } + @Test + public void metadataRequestIdTakesPrecedenceOverAmbientRequestId() { + NexusOperationMetadata metadata = + new NexusOperationMetadata( + "nexus-request-id", "http://localhost/callback", Collections.emptyMap()); + nexusContext.setNexusOperationMetadata(metadata); + + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + Assert.assertEquals("nexus-request-id", captor.getValue().getRequestId()); + } + @Test public void nexusMetadataWithEmptyCallbackUrlOmitsCompletionCallback() { NexusOperationMetadata metadata = @@ -132,8 +149,61 @@ public void nexusMetadataWithEmptyCallbackUrlOmitsCompletionCallback() { } @Test - public void nexusContextWithoutMetadataStartsOrdinaryActivity() { - nexusContext.setRequestLinks(Collections.singletonList(workflowEventLink())); + public void nexusContextWithoutMetadataGetsAmbientLinksAndAmbientRequestIdButNoCallback() { + Link link = workflowEventLink(); + nexusContext.setRequestLinks(Collections.singletonList(link)); + nexusContext.setRequestId("ambient-nexus-request-id"); + + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + StartActivityExecutionRequest request = captor.getValue(); + Assert.assertEquals("ambient-nexus-request-id", request.getRequestId()); + Assert.assertEquals(Collections.singletonList(link), request.getLinksList()); + Assert.assertEquals(0, request.getCompletionCallbacksCount()); + Assert.assertTrue(request.getOnConflictOptions().getAttachRequestId()); + Assert.assertTrue(request.getOnConflictOptions().getAttachLinks()); + Assert.assertFalse(request.getOnConflictOptions().getAttachCompletionCallbacks()); + Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks()); + } + + @Test + public void twoStartsInTheSameInvocationShareTheAmbientRequestId() { + nexusContext.setRequestId("ambient-nexus-request-id"); + + invoker.startActivity(newStartActivityInput()); + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient, times(2)).startActivity(captor.capture()); + List requests = captor.getAllValues(); + Assert.assertEquals("ambient-nexus-request-id", requests.get(0).getRequestId()); + Assert.assertEquals("ambient-nexus-request-id", requests.get(1).getRequestId()); + } + + @Test + public void nexusContextWithoutAmbientStateStartsOrdinaryActivity() { + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + StartActivityExecutionRequest request = captor.getValue(); + Assert.assertFalse(request.getRequestId().isEmpty()); + Assert.assertEquals(0, request.getLinksCount()); + Assert.assertEquals(0, request.getCompletionCallbacksCount()); + Assert.assertTrue(request.getOnConflictOptions().getAttachRequestId()); + Assert.assertTrue(request.getOnConflictOptions().getAttachLinks()); + Assert.assertFalse(request.getOnConflictOptions().getAttachCompletionCallbacks()); + Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks()); + } + + @Test + public void outsideNexusContextStartsOrdinaryActivity() { + CurrentNexusOperationContext.unset(); invoker.startActivity(newStartActivityInput()); diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/nexus/ActivityOperationLinkingTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/ActivityOperationLinkingTest.java new file mode 100644 index 0000000000..aa4edba0a5 --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/workflow/nexus/ActivityOperationLinkingTest.java @@ -0,0 +1,191 @@ +package io.temporal.workflow.nexus; + +import static io.temporal.internal.common.WorkflowExecutionUtils.getEventOfType; +import static org.junit.Assume.assumeTrue; + +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.common.v1.Link; +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.history.v1.History; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.client.ActivityClient; +import io.temporal.client.ActivityClientOptions; +import io.temporal.client.ActivityExecutionDescription; +import io.temporal.client.ActivityHandle; +import io.temporal.client.StartActivityOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.nexus.Nexus; +import io.temporal.testing.internal.SDKTestWorkflowRule; +import io.temporal.workflow.NexusOperationOptions; +import io.temporal.workflow.NexusServiceOptions; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.shared.TestNexusServices; +import io.temporal.workflow.shared.TestWorkflows; +import java.time.Duration; +import java.util.HashSet; +import java.util.Set; +import java.util.UUID; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; + +/** + * Verifies link propagation with activities when a synchronous Nexus operation handler starts more + * than one activity via a raw {@link ActivityClient} obtained from {@link + * Nexus#getOperationContext()}. + * + *
    + *
  • Forward direction: each activity's own record links back to the caller's {@code + * NexusOperationScheduled} event. + *
  • Backward direction: both activities' completions land as response links on the caller's + * single {@code NexusOperationCompleted} event. + *
+ * + *

Requires a real server; the in-process test server does not implement {@code + * StartActivityExecution} (see {@link AsyncActivityOperationTest}, which has the same gate). + */ +public class ActivityOperationLinkingTest { + + @Rule + public SDKTestWorkflowRule testWorkflowRule = + SDKTestWorkflowRule.newBuilder() + .setWorkflowTypes(TestNexus.class) + .setActivityImplementations(new TestActivityImpl()) + .setNexusServiceImplementation(new TestNexusServiceImpl()) + .build(); + + @BeforeClass + public static void requireExternalService() { + assumeTrue( + "standalone-activity Nexus links require a real server", + SDKTestWorkflowRule.useExternalService); + } + + @Test + public void testTwoActivitiesBothLinkToOperation() { + String input = "world-" + UUID.randomUUID(); + TestWorkflows.TestWorkflow1 workflowStub = + testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class); + String result = workflowStub.execute(input); + Assert.assertEquals("hello " + input + "-a|hello " + input + "-b", result); + + String callerWorkflowId = WorkflowStub.fromTyped(workflowStub).getExecution().getWorkflowId(); + History callerHistory = + testWorkflowRule.getWorkflowClient().fetchHistory(callerWorkflowId).getHistory(); + + // Backward direction: both activities' completions must land on the caller's single + // NexusOperationCompleted event as response links, not just the guarded/first one. + HistoryEvent completed = + getEventOfType(callerHistory, EventType.EVENT_TYPE_NEXUS_OPERATION_COMPLETED); + Assert.assertNotNull("expected a NexusOperationCompleted event", completed); + Assert.assertEquals("expected one response link per activity", 2, completed.getLinksCount()); + Set linkedActivityIds = new HashSet<>(); + for (int i = 0; i < completed.getLinksCount(); i++) { + Link.Activity activityLink = completed.getLinks(i).getActivity(); + Assert.assertNotNull("expected an Activity-typed response link", activityLink); + linkedActivityIds.add(activityLink.getActivityId()); + } + Assert.assertTrue(linkedActivityIds.contains("act-" + input + "-a")); + Assert.assertTrue(linkedActivityIds.contains("act-" + input + "-b")); + + // Forward direction: each activity's own record links back to the caller's + // NexusOperationScheduled event, not just the guarded/first one. + ActivityClient activityClient = + ActivityClient.newInstance( + testWorkflowRule.getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder().setNamespace(SDKTestWorkflowRule.NAMESPACE).build()); + for (String suffix : new String[] {"a", "b"}) { + String activityId = "act-" + input + "-" + suffix; + ActivityExecutionDescription description = + activityClient.getHandle(activityId, null).describe(); + Assert.assertTrue( + "expected at least one link on activity " + activityId, + description.getRawInfo().getLinksCount() >= 1); + Link.WorkflowEvent forwardLink = description.getRawInfo().getLinks(0).getWorkflowEvent(); + Assert.assertNotNull( + "expected a WorkflowEvent-typed forward link on activity " + activityId, forwardLink); + Assert.assertEquals(callerWorkflowId, forwardLink.getWorkflowId()); + Assert.assertEquals( + EventType.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, forwardLink.getEventRef().getEventType()); + } + } + + public static class TestNexus implements TestWorkflows.TestWorkflow1 { + @Override + public String execute(String input) { + NexusServiceOptions serviceOptions = + NexusServiceOptions.newBuilder() + .setOperationOptions( + NexusOperationOptions.newBuilder() + .setScheduleToCloseTimeout(Duration.ofSeconds(30)) + .build()) + .build(); + TestNexusServices.TestNexusService1 stub = + Workflow.newNexusServiceStub(TestNexusServices.TestNexusService1.class, serviceOptions); + return stub.operation(input); + } + } + + @ActivityInterface + public interface TestActivity { + @ActivityMethod + String process(String input); + } + + public static class TestActivityImpl implements TestActivity { + @Override + public String process(String input) { + return "hello " + input; + } + } + + /** + * Starts two activities inline via a raw {@link ActivityClient} obtained from {@link + * Nexus#getOperationContext()} instead of {@code TemporalOperationHandler}'s single-guarded-call + * {@code TemporalNexusClient} -- the only way to start more than one activity synchronously in + * one Nexus operation invocation. + */ + @ServiceImpl(service = TestNexusServices.TestNexusService1.class) + public class TestNexusServiceImpl { + @OperationImpl + public OperationHandler operation() { + return OperationHandler.sync( + (ctx, details, input) -> { + ActivityClient activityClient = + ActivityClient.newInstance( + Nexus.getOperationContext().getWorkflowClient().getWorkflowServiceStubs(), + ActivityClientOptions.newBuilder() + .setNamespace(Nexus.getOperationContext().getInfo().getNamespace()) + .build()); + String taskQueue = Nexus.getOperationContext().getInfo().getTaskQueue(); + + ActivityHandle first = + activityClient.start( + TestActivity.class, + TestActivity::process, + StartActivityOptions.newBuilder() + .setId("act-" + input + "-a") + .setTaskQueue(taskQueue) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(), + input + "-a"); + ActivityHandle second = + activityClient.start( + TestActivity.class, + TestActivity::process, + StartActivityOptions.newBuilder() + .setId("act-" + input + "-b") + .setTaskQueue(taskQueue) + .setStartToCloseTimeout(Duration.ofSeconds(10)) + .build(), + input + "-b"); + return first.getResult() + "|" + second.getResult(); + }); + } + } +} From 2764436415f89701c05af89dfe6fcebe661fc26a Mon Sep 17 00:00:00 2001 From: tekkaya <86028633+tekkaya@users.noreply.github.com> Date: Mon, 31 Aug 2026 16:56:17 -0700 Subject: [PATCH 2/2] Only set OnConflictOptions when there is a link or callback to attach The server rejects a StartActivityExecutionRequest whose OnConflictOptions sets attach_request_id when the request carries neither a link nor a completion callback (chasm/lib/activity/validator.go's validateOnConflictOptions: "attach_request_id requires at least one completion callback or link"). The previous code set attach_request_id and attach_links unconditionally whenever any Nexus context existed, regardless of whether the inbound task actually had links -- a bypass-path activity start issued during an invocation whose inbound Nexus task carries no links would send exactly that invalid combination and get rejected. This wasn't caught before because the existing unit tests mock the client and never exercise real server-side validation. OnConflictOptions is now only set when there's something to attach, and each flag reflects what the request actually carries, matching sdk-go's identical gating in its own fix (temporalnexus/temporal_operation.go, PR #2633). This also fixes the guarded (metadata-backed) path: it previously set attach_completion_callbacks based on whether metadata was present rather than whether a callback URL was actually set, so a guarded start with an empty callback URL and no links would hit the same rejection. RootActivityClientInvokerTest: flipped the assertion in nexusMetadataWithEmptyCallbackUrlOmitsCompletionCallback (attach_completion_ callbacks now correctly reflects the absence of a real callback), rewrote nexusContextWithoutAmbientStateStartsOrdinaryActivity to assert OnConflictOptions is entirely absent, and added metadataWithEmptyCallbackUrlAndNoLinksOmitsOnConflictOptions covering the previously-untested guarded-call variant of the same bug. --- .../client/RootActivityClientInvoker.java | 19 ++++++++------ .../client/RootActivityClientInvokerTest.java | 25 ++++++++++++++++--- 2 files changed, 33 insertions(+), 11 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java index 902a847b2c..c2598fc517 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java @@ -136,14 +136,19 @@ public StartActivityOutput startActivity(StartActivityInput input) { // completes the Nexus operation. protoLinks = nexusContext.getRequestLinks(); request.addAllLinks(protoLinks); - io.temporal.api.common.v1.OnConflictOptions.Builder onConflictOptions = + } + + boolean willAttachCompletionCallback = + nexusOperationMetadata != null + && !Strings.isNullOrEmpty(nexusOperationMetadata.callbackUrl); + if (nexusContext != null && (!protoLinks.isEmpty() || willAttachCompletionCallback)) { + // The server rejects attach_request_id unless the request also carries at least one link + // or completion callback to attach on conflict. + request.setOnConflictOptions( io.temporal.api.common.v1.OnConflictOptions.newBuilder() .setAttachRequestId(true) - .setAttachLinks(true); - if (nexusOperationMetadata != null) { - onConflictOptions.setAttachCompletionCallbacks(true); - } - request.setOnConflictOptions(onConflictOptions); + .setAttachLinks(!protoLinks.isEmpty()) + .setAttachCompletionCallbacks(willAttachCompletionCallback)); } if (nexusOperationMetadata != null) { @@ -159,7 +164,7 @@ public StartActivityOutput startActivity(StartActivityInput input) { "failed to generate activity operation token", e); } - if (!Strings.isNullOrEmpty(nexusOperationMetadata.callbackUrl)) { + if (willAttachCompletionCallback) { Callback cb = InternalUtils.buildNexusCallback( nexusOperationMetadata.callbackUrl, diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java index 28bfe74593..aaffe60c9d 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/RootActivityClientInvokerTest.java @@ -143,11 +143,30 @@ public void nexusMetadataWithEmptyCallbackUrlOmitsCompletionCallback() { Assert.assertEquals(0, request.getCompletionCallbacksCount()); Assert.assertTrue(request.getOnConflictOptions().getAttachRequestId()); Assert.assertTrue(request.getOnConflictOptions().getAttachLinks()); - Assert.assertTrue(request.getOnConflictOptions().getAttachCompletionCallbacks()); + Assert.assertFalse(request.getOnConflictOptions().getAttachCompletionCallbacks()); Assert.assertNotNull(metadata.operationToken); Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks()); } + @Test + public void metadataWithEmptyCallbackUrlAndNoLinksOmitsOnConflictOptions() { + NexusOperationMetadata metadata = + new NexusOperationMetadata( + "nexus-request-id", "", Collections.singletonMap("Custom-Header", "value")); + nexusContext.setNexusOperationMetadata(metadata); + + invoker.startActivity(newStartActivityInput()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(StartActivityExecutionRequest.class); + verify(genericClient).startActivity(captor.capture()); + StartActivityExecutionRequest request = captor.getValue(); + Assert.assertEquals("nexus-request-id", request.getRequestId()); + Assert.assertEquals(0, request.getLinksCount()); + Assert.assertEquals(0, request.getCompletionCallbacksCount()); + Assert.assertFalse(request.hasOnConflictOptions()); + } + @Test public void nexusContextWithoutMetadataGetsAmbientLinksAndAmbientRequestIdButNoCallback() { Link link = workflowEventLink(); @@ -195,9 +214,7 @@ public void nexusContextWithoutAmbientStateStartsOrdinaryActivity() { Assert.assertFalse(request.getRequestId().isEmpty()); Assert.assertEquals(0, request.getLinksCount()); Assert.assertEquals(0, request.getCompletionCallbacksCount()); - Assert.assertTrue(request.getOnConflictOptions().getAttachRequestId()); - Assert.assertTrue(request.getOnConflictOptions().getAttachLinks()); - Assert.assertFalse(request.getOnConflictOptions().getAttachCompletionCallbacks()); + Assert.assertFalse(request.hasOnConflictOptions()); Assert.assertEquals(Collections.singletonList(activityLink()), nexusContext.getResponseLinks()); }