22
33import static org .junit .Assume .assumeTrue ;
44
5- import com .google .protobuf .util .Durations ;
6- import io .grpc .Status ;
7- import io .grpc .StatusRuntimeException ;
85import io .nexusrpc .handler .OperationCancelDetails ;
96import io .nexusrpc .handler .OperationContext ;
107import io .nexusrpc .handler .OperationHandler ;
1512import io .temporal .api .enums .v1 .EventType ;
1613import io .temporal .api .history .v1 .History ;
1714import io .temporal .api .history .v1 .HistoryEvent ;
18- import io .temporal .api .workflowservice .v1 .RegisterNamespaceRequest ;
1915import io .temporal .client .BatchRequest ;
2016import io .temporal .client .WorkflowClient ;
21- import io .temporal .client .WorkflowClientOptions ;
2217import io .temporal .client .WorkflowOptions ;
2318import io .temporal .client .WorkflowStub ;
2419import io .temporal .nexus .Nexus ;
25- import io .temporal .serviceclient .WorkflowServiceStubs ;
26- import io .temporal .serviceclient .WorkflowServiceStubsOptions ;
2720import io .temporal .testing .internal .SDKTestWorkflowRule ;
2821import io .temporal .workflow .NexusOperationHandle ;
2922import io .temporal .workflow .NexusOperationOptions ;
4033import java .util .List ;
4134import java .util .UUID ;
4235import javax .annotation .Nullable ;
43- import org .junit .After ;
4436import org .junit .Assert ;
4537import org .junit .BeforeClass ;
4638import org .junit .Rule ;
4739import org .junit .Test ;
4840
4941/**
5042 * Verifies link propagation in both directions when a Nexus operation handler interacts with a
51- * workflow via signal. Covers four scenarios:
43+ * workflow via signal. Covers three scenarios:
5244 *
5345 * <ul>
54- * <li>{@link #testSignalOperationLinks()} — same-namespace, sync handler, two signals
55- * (signalWithStart + plain signal).
56- * <li>{@link #testCrossNamespaceSignalOperationLinks()} — caller and callee in different
57- * namespaces; otherwise identical to the same-namespace case.
46+ * <li>{@link #testSignalOperationLinks()} — sync handler, two signals (signalWithStart + plain
47+ * signal).
5848 * <li>{@link #testMultiSignalOperationLinks()} — one Nexus operation signals three different
5949 * callees; verifies all three backlinks land on the caller's single {@code
6050 * NexusOperationCompleted} event.
6353 * {@link io.temporal.internal.nexus.NexusTaskHandlerImpl}).
6454 * </ul>
6555 *
66- * <p>All four tests require Temporal server ≥ 1.31 with {@code EnableCHASMSignalBacklinks=true};
67- * the in-memory test server does not implement this path so the class is skipped unless a real
68- * server is in use.
56+ * <p>All tests require Temporal server ≥ 1.31 with {@code EnableCHASMSignalBacklinks=true}; the
57+ * in-memory test server does not implement this path so the class is skipped unless a real server
58+ * is in use.
6959 */
7060public class SignalOperationLinkingTest extends BaseNexusTest {
7161
7262 private static final String MODE_SIGNAL_WITH_START = "signalWithStart" ;
7363 private static final String MODE_SIGNAL = "signal" ;
7464 private static final String MODE_MULTI_SIGNAL_WITH_START = "multi" ;
7565 private static final String MODE_ASYNC_SIGNAL_WITH_START = "asyncSignalWithStart" ;
76- private static final String CALLEE_NAMESPACE = "UnitTest2" ;
7766
78- // Caller workflow + Nexus handler register here (namespace UnitTest).
7967 @ Rule
8068 public SDKTestWorkflowRule testWorkflowRule =
8169 SDKTestWorkflowRule .newBuilder ()
8270 .setWorkflowTypes (SignalCallerWorkflow .class , SignalCalleeWorkflowImpl .class )
8371 .setNexusServiceImplementation (new SignalingNexusServiceImpl ())
8472 .build ();
8573
86- // Separate worker/client on the callee namespace, used by the cross-namespace test. No
87- // precedent in the repo for multi-@Rule SDKTestWorkflowRule patterns; every test method pays
88- // the cost of starting this second worker even if it doesn't use it. Acceptable for the
89- // current test count; revisit if more cross-namespace tests get added.
90- @ Rule
91- public SDKTestWorkflowRule calleeNamespaceRule =
92- SDKTestWorkflowRule .newBuilder ()
93- .setNamespace (CALLEE_NAMESPACE )
94- .setWorkflowTypes (SignalCalleeWorkflowImpl .class )
95- .build ();
96-
9774 @ BeforeClass
98- public static void requireExternalServiceAndSetupCalleeNamespace () {
75+ public static void requireExternalService () {
9976 // The server-side backlink implementation (temporalio/temporal#9897) is gated by
10077 // EnableCHASMSignalBacklinks and is only present in real servers.
10178 assumeTrue (
10279 "signal backlinks require a real server with EnableCHASMSignalBacklinks=true" ,
10380 SDKTestWorkflowRule .useExternalService );
104- // The test rule does not auto-register namespaces on an external server.
105- ensureNamespaceExists (CALLEE_NAMESPACE );
106- }
107-
108- @ After
109- public void resetNamespaceOverrides () {
110- SignalingNexusServiceImpl .calleeNamespaceOverride = null ;
111- SignalingNexusServiceImpl .calleeTaskQueueOverride = null ;
11281 }
11382
11483 @ Override
@@ -120,14 +89,7 @@ protected SDKTestWorkflowRule getTestWorkflowRule() {
12089
12190 @ Test
12291 public void testSignalOperationLinks () {
123- runTwoSignalScenario (testWorkflowRule );
124- }
125-
126- @ Test
127- public void testCrossNamespaceSignalOperationLinks () {
128- SignalingNexusServiceImpl .calleeNamespaceOverride = CALLEE_NAMESPACE ;
129- SignalingNexusServiceImpl .calleeTaskQueueOverride = calleeNamespaceRule .getTaskQueue ();
130- runTwoSignalScenario (calleeNamespaceRule );
92+ runTwoSignalScenario ();
13193 }
13294
13395 /**
@@ -224,28 +186,22 @@ public void testAsyncSignalOperationLinks() {
224186
225187 // ── Shared scenario + assertion helpers ──────────────────────────────────────────────────
226188
227- /**
228- * Drive the two-signal flow (signalWithStart + plain signal) and assert link propagation. Used by
229- * same-namespace and cross-namespace tests; the only thing that varies is which rule's client
230- * fetches the callee history.
231- */
232- private void runTwoSignalScenario (SDKTestWorkflowRule calleeRule ) {
233- WorkflowClient callerClient = testWorkflowRule .getWorkflowClient ();
234- WorkflowClient calleeClient = calleeRule .getWorkflowClient ();
189+ /** Drive the two-signal flow (signalWithStart + plain signal) and assert link propagation. */
190+ private void runTwoSignalScenario () {
191+ WorkflowClient client = testWorkflowRule .getWorkflowClient ();
235192 String calleeWorkflowId = "signal-callee-" + UUID .randomUUID ();
236193
237194 TestWorkflows .TestWorkflow1 callerStub =
238195 testWorkflowRule .newWorkflowStubTimeoutOptions (TestWorkflows .TestWorkflow1 .class , "caller" );
239196 String result = callerStub .execute ("twoSync:" + calleeWorkflowId );
240197 Assert .assertEquals ("ok:signalWithStart|ok:signal" , result );
241198
242- String calleeResult =
243- calleeClient .newUntypedWorkflowStub (calleeWorkflowId ).getResult (String .class );
199+ String calleeResult = client .newUntypedWorkflowStub (calleeWorkflowId ).getResult (String .class );
244200 Assert .assertEquals ("first,second" , calleeResult );
245201
246202 String callerWorkflowId = WorkflowStub .fromTyped (callerStub ).getExecution ().getWorkflowId ();
247- History callerHistory = callerClient .fetchHistory (callerWorkflowId ).getHistory ();
248- History calleeHistory = calleeClient .fetchHistory (calleeWorkflowId ).getHistory ();
203+ History callerHistory = client .fetchHistory (callerWorkflowId ).getHistory ();
204+ History calleeHistory = client .fetchHistory (calleeWorkflowId ).getHistory ();
249205
250206 assertForwardLinks (calleeHistory , callerWorkflowId , /* expectedCount= */ 2 );
251207
@@ -314,36 +270,6 @@ private static List<HistoryEvent> getAllEventsOfType(History history, EventType
314270 return out ;
315271 }
316272
317- /**
318- * Register {@code namespace} on the external server if it doesn't already exist. Honours the
319- * {@code TEMPORAL_SERVICE_ADDRESS} env var the same way {@code
320- * io.temporal.testing.internal.ExternalServiceTestConfigurator} does, so the test works against
321- * whichever server the test rule itself connects to.
322- */
323- private static void ensureNamespaceExists (String namespace ) {
324- String target = System .getenv ("TEMPORAL_SERVICE_ADDRESS" );
325- WorkflowServiceStubsOptions .Builder optionsBuilder = WorkflowServiceStubsOptions .newBuilder ();
326- if (target != null && !target .isEmpty ()) {
327- optionsBuilder .setTarget (target );
328- }
329- WorkflowServiceStubs stubs = WorkflowServiceStubs .newServiceStubs (optionsBuilder .build ());
330- try {
331- stubs
332- .blockingStub ()
333- .registerNamespace (
334- RegisterNamespaceRequest .newBuilder ()
335- .setNamespace (namespace )
336- .setWorkflowExecutionRetentionPeriod (Durations .fromHours (24 ))
337- .build ());
338- } catch (StatusRuntimeException e ) {
339- if (e .getStatus ().getCode () != Status .Code .ALREADY_EXISTS ) {
340- throw e ;
341- }
342- } finally {
343- stubs .shutdownNow ();
344- }
345- }
346-
347273 // ── Workflows ────────────────────────────────────────────────────────────────────────────
348274
349275 /**
@@ -431,12 +357,10 @@ public void ping(String msg) {
431357
432358 /**
433359 * Single Nexus operation that dispatches based on a mode prefix in its input. Supports sync and
434- * async return shapes and an optional namespace override for cross-namespace tests .
360+ * async return shapes.
435361 */
436362 @ ServiceImpl (service = TestNexusServices .TestNexusService1 .class )
437363 public static class SignalingNexusServiceImpl {
438- static volatile String calleeNamespaceOverride ;
439- static volatile String calleeTaskQueueOverride ;
440364
441365 @ OperationImpl
442366 public OperationHandler <String , String > operation () {
@@ -449,36 +373,23 @@ public OperationStartResult<String> start(
449373 String rest = parts [1 ];
450374
451375 io .temporal .nexus .NexusOperationContext opCtx = Nexus .getOperationContext ();
452- WorkflowClient ambient = opCtx .getWorkflowClient ();
453- WorkflowClient calleeClient =
454- calleeNamespaceOverride == null
455- ? ambient
456- : WorkflowClient .newInstance (
457- ambient .getWorkflowServiceStubs (),
458- WorkflowClientOptions .newBuilder ()
459- .setNamespace (calleeNamespaceOverride )
460- .build ());
461- String taskQueue =
462- calleeTaskQueueOverride != null
463- ? calleeTaskQueueOverride
464- : opCtx .getInfo ().getTaskQueue ();
376+ WorkflowClient client = opCtx .getWorkflowClient ();
377+ String taskQueue = opCtx .getInfo ().getTaskQueue ();
465378
466379 switch (mode ) {
467380 case MODE_SIGNAL_WITH_START :
468- signalWithStart (calleeClient , rest , taskQueue , /* expectedSignals= */ 2 , "first" );
381+ signalWithStart (client , rest , taskQueue , /* expectedSignals= */ 2 , "first" );
469382 return OperationStartResult .sync ("ok:" + MODE_SIGNAL_WITH_START );
470383 case MODE_SIGNAL :
471- calleeClient .newWorkflowStub (SignalCalleeWorkflow .class , rest ).ping ("second" );
384+ client .newWorkflowStub (SignalCalleeWorkflow .class , rest ).ping ("second" );
472385 return OperationStartResult .sync ("ok:" + MODE_SIGNAL );
473386 case MODE_MULTI_SIGNAL_WITH_START :
474387 for (String id : rest .split ("," )) {
475- signalWithStart (
476- calleeClient , id , taskQueue , /* expectedSignals= */ 1 , "multi-signal" );
388+ signalWithStart (client , id , taskQueue , /* expectedSignals= */ 1 , "multi-signal" );
477389 }
478390 return OperationStartResult .sync ("ok:multi:" + rest );
479391 case MODE_ASYNC_SIGNAL_WITH_START :
480- signalWithStart (
481- calleeClient , rest , taskQueue , /* expectedSignals= */ 1 , "async-signal" );
392+ signalWithStart (client , rest , taskQueue , /* expectedSignals= */ 1 , "async-signal" );
482393 // Async branch in NexusTaskHandlerImpl. The caller never waits for completion, so
483394 // the token is opaque.
484395 return OperationStartResult .async ("async-op-" + UUID .randomUUID ());
0 commit comments