From 20c2e5014bf81e7872918b1f9bf57f4315c9449f Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Fri, 13 Jun 2025 16:41:53 -0700 Subject: [PATCH 1/2] Add CancellationType to nexus cancellation sample --- .../samples/nexuscancellation/README.MD | 19 +++++++-- .../caller/HelloCallerWorkflowImpl.java | 9 +++- .../handler/HelloHandlerWorkflowImpl.java | 41 ++++++++++++------- 3 files changed, 50 insertions(+), 19 deletions(-) diff --git a/core/src/main/java/io/temporal/samples/nexuscancellation/README.MD b/core/src/main/java/io/temporal/samples/nexuscancellation/README.MD index f4da82852..bd69da875 100644 --- a/core/src/main/java/io/temporal/samples/nexuscancellation/README.MD +++ b/core/src/main/java/io/temporal/samples/nexuscancellation/README.MD @@ -1,6 +1,6 @@ # Nexus Cancellation -This sample shows how to cancel a Nexus operation from a caller workflow. +This sample shows how to cancel a Nexus operation from a caller workflow and specify a cancellation type. In this sample we will show using the `WAIT_REQUESTED` cancellation type, which allows the caller to return after the handler workflow has received the requested to be cancelled, but does not wait for the handler workflow to finish processing the cancellation request. To run this sample, set up your environment following the instructions in the main [Nexus Sample](../nexus/README.md). @@ -29,8 +29,19 @@ Next, in separate terminal windows: ### Output -which should result in: +which should result in on the caller side: ``` -INFO i.t.s.n.caller.CallerStarter - Started workflow workflowId: 326732dd-a2b1-4de7-9ddd-dcee4f9f0229 runId: d580499f-79d5-461d-bd49-6248b4e522ae -INFO i.t.s.n.caller.CallerStarter - Workflow result: Hallo Nexus πŸ‘‹ +14:33:52.810 i.t.s.n.caller.CallerStarter - Started workflow workflowId: 87e97bf0-ca8a-4ae6-a9dc-ae97e5c0ac41 runId: 01976b36-a524-71a1-b848-8eb385fec2c3 +14:33:54.250 i.t.s.n.caller.CallerStarter - Workflow result: Hallo Nexus πŸ‘‹ ``` + +on the handler side: + +``` +14:33:54.177 INFO i.t.s.n.h.HelloHandlerWorkflowImpl - HelloHandlerWorkflow was cancelled successfully. +14:33:56.167 INFO i.t.s.n.h.HelloHandlerWorkflowImpl - HelloHandlerWorkflow was cancelled successfully. +14:33:57.172 INFO i.t.s.n.h.HelloHandlerWorkflowImpl - HelloHandlerWorkflow was cancelled successfully. +14:33:57.176 INFO i.t.s.n.h.HelloHandlerWorkflowImpl - HelloHandlerWorkflow was cancelled successfully. +``` + +Notice the timing, the caller workflow returned before the handler workflow was cancelled. This is because of the use of `WAIT_REQUESTED` as the cancellation type in the Nexus operation. This means the caller didn't have to wait for the handler workflow to finish, but still guarantees the handler workflow will receive the cancellation request. \ No newline at end of file diff --git a/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java b/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java index c3dc6440c..e74971693 100644 --- a/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java +++ b/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java @@ -22,6 +22,12 @@ public class HelloCallerWorkflowImpl implements HelloCallerWorkflow { .setOperationOptions( NexusOperationOptions.newBuilder() .setScheduleToCloseTimeout(Duration.ofSeconds(10)) + // Set the cancellation type to WAIT_REQUESTED. This means that the caller + // will wait for the cancellation request to be received by the handler before + // proceeding with the cancellation. + // + // By default, the caller would wait until the operation is completed. + .setCancellationType(NexusOperationCancellationType.WAIT_REQUESTED) .build()) .build()); @@ -55,7 +61,8 @@ public String hello(String message) { // Trigger cancellation of all uncompleted nexus operations invocations within the cancellation // scope scope.cancel(); - // Optionally, wait for all nexus operations to complete + // Optionally, wait for all nexus operations to receive a cancellation request before + // proceeding. // // Note: Once the workflow completes any pending cancellation requests are dropped by the // server. diff --git a/core/src/main/java/io/temporal/samples/nexuscancellation/handler/HelloHandlerWorkflowImpl.java b/core/src/main/java/io/temporal/samples/nexuscancellation/handler/HelloHandlerWorkflowImpl.java index 406b4b20b..ca6510f60 100644 --- a/core/src/main/java/io/temporal/samples/nexuscancellation/handler/HelloHandlerWorkflowImpl.java +++ b/core/src/main/java/io/temporal/samples/nexuscancellation/handler/HelloHandlerWorkflowImpl.java @@ -1,29 +1,42 @@ package io.temporal.samples.nexuscancellation.handler; import io.temporal.failure.ApplicationFailure; +import io.temporal.failure.CanceledFailure; import io.temporal.samples.nexus.handler.HelloHandlerWorkflow; import io.temporal.samples.nexus.service.NexusService; import io.temporal.workflow.Workflow; import java.time.Duration; +import org.slf4j.Logger; public class HelloHandlerWorkflowImpl implements HelloHandlerWorkflow { + public static final Logger log = Workflow.getLogger(HelloHandlerWorkflowImpl.class); + @Override public NexusService.HelloOutput hello(NexusService.HelloInput input) { // Sleep for a random duration to simulate some work - Workflow.sleep(Duration.ofSeconds(Workflow.newRandom().nextInt(5))); - switch (input.getLanguage()) { - case EN: - return new NexusService.HelloOutput("Hello " + input.getName() + " πŸ‘‹"); - case FR: - return new NexusService.HelloOutput("Bonjour " + input.getName() + " πŸ‘‹"); - case DE: - return new NexusService.HelloOutput("Hallo " + input.getName() + " πŸ‘‹"); - case ES: - return new NexusService.HelloOutput("Β‘Hola! " + input.getName() + " πŸ‘‹"); - case TR: - return new NexusService.HelloOutput("Merhaba " + input.getName() + " πŸ‘‹"); + try { + Workflow.sleep(Duration.ofSeconds(Workflow.newRandom().nextInt(5))); + switch (input.getLanguage()) { + case EN: + return new NexusService.HelloOutput("Hello " + input.getName() + " πŸ‘‹"); + case FR: + return new NexusService.HelloOutput("Bonjour " + input.getName() + " πŸ‘‹"); + case DE: + return new NexusService.HelloOutput("Hallo " + input.getName() + " πŸ‘‹"); + case ES: + return new NexusService.HelloOutput("Β‘Hola! " + input.getName() + " πŸ‘‹"); + case TR: + return new NexusService.HelloOutput("Merhaba " + input.getName() + " πŸ‘‹"); + } + throw ApplicationFailure.newFailure( + "Unsupported language: " + input.getLanguage(), "UNSUPPORTED_LANGUAGE"); + } catch (CanceledFailure e) { + // Simulate some work after cancellation is requested + Workflow.newDetachedCancellationScope( + () -> Workflow.sleep(Duration.ofSeconds(Workflow.newRandom().nextInt(5)))) + .run(); + log.info("HelloHandlerWorkflow was cancelled successfully."); + throw e; } - throw ApplicationFailure.newFailure( - "Unsupported language: " + input.getLanguage(), "UNSUPPORTED_LANGUAGE"); } } From c1aaaea68a38bbaa33ca7652fdeb1ee6c9da7208 Mon Sep 17 00:00:00 2001 From: Quinn Klassen Date: Tue, 24 Jun 2025 16:51:29 -0700 Subject: [PATCH 2/2] Clarify --- .../nexuscancellation/caller/HelloCallerWorkflowImpl.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java b/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java index e74971693..1bea801c8 100644 --- a/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java +++ b/core/src/main/java/io/temporal/samples/nexuscancellation/caller/HelloCallerWorkflowImpl.java @@ -61,11 +61,12 @@ public String hello(String message) { // Trigger cancellation of all uncompleted nexus operations invocations within the cancellation // scope scope.cancel(); - // Optionally, wait for all nexus operations to receive a cancellation request before + // Wait for all nexus operations to receive a cancellation request before // proceeding. // // Note: Once the workflow completes any pending cancellation requests are dropped by the - // server. + // server. In general, it is a good practice to wait for all cancellation requests to be + // processed before completing the workflow. for (Promise promise : results) { try { promise.get();