From 6904f118e4638152c661ef60bd486bb3f52af666 Mon Sep 17 00:00:00 2001 From: Mana Agrawal Date: Thu, 6 Aug 2026 13:21:00 -0700 Subject: [PATCH] Add Resilient Fan-Out sample: partial-failure tolerance in a heterogeneous DAG Adds a sample demonstrating a pattern not covered by the existing Child Workflow docs/samples: a fan-out with one critical branch and several best-effort branches, where: - each branch is awaited independently (not via Promise.allOf(...)), so a best-effort failure can't take down a healthy critical branch or vice versa - still-running best-effort children are explicitly cancelled via a CancellationScope when the critical branch fails, instead of relying on the default ParentClosePolicy (TERMINATE), which kills a child with no CanceledFailure to react to - failure is reconciled via Saga with parallel compensation, repurposed for independent, isolated notifications rather than classic "undo what I did" semantics Includes a runnable Starter for all three outcomes and unit tests covering the happy path, a degraded-but-successful run, and a critical failure that triggers cancellation + reconciliation. --- README.md | 2 + .../CreateAccountWorkflow.java | 12 ++ .../CreateAccountWorkflowImpl.java | 18 +++ .../resilientfanout/OnboardingActivities.java | 17 +++ .../OnboardingActivitiesImpl.java | 42 +++++++ .../resilientfanout/OnboardingResult.java | 48 ++++++++ .../resilientfanout/OnboardingWorkflow.java | 11 ++ .../OnboardingWorkflowImpl.java | 112 ++++++++++++++++++ .../ProvisionStorageWorkflow.java | 12 ++ .../ProvisionStorageWorkflowImpl.java | 18 +++ .../samples/resilientfanout/README.md | 45 +++++++ .../ReconciliationActivities.java | 19 +++ .../ReconciliationActivitiesImpl.java | 19 +++ .../SendWelcomeEmailWorkflow.java | 12 ++ .../SendWelcomeEmailWorkflowImpl.java | 18 +++ .../samples/resilientfanout/Starter.java | 58 +++++++++ .../OnboardingWorkflowTest.java | 96 +++++++++++++++ 17 files changed, 559 insertions(+) create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflow.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflowImpl.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivities.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivitiesImpl.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/OnboardingResult.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflow.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflowImpl.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflow.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflowImpl.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/README.md create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivities.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivitiesImpl.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflow.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflowImpl.java create mode 100644 core/src/main/java/io/temporal/samples/resilientfanout/Starter.java create mode 100644 core/src/test/java/io/temporal/samples/resilientfanout/OnboardingWorkflowTest.java diff --git a/README.md b/README.md index 1465a9cc..30745b50 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,8 @@ See the README.md file in each main sample directory for cut/paste Gradle comman - [**Synchronous Booking SAGA**](/core/src/main/java/io/temporal/samples/bookingsyncsaga): Demonstrates low latency SAGA with potentially long compensations. +- [**Resilient Fan-Out**](/core/src/main/java/io/temporal/samples/resilientfanout): Demonstrates a heterogeneous fan-out (a critical branch plus best-effort branches) with partial-failure tolerance: awaiting each Child Workflow independently instead of failing fast on the first one, cancelling still-running best-effort children when the critical branch fails, and reconciling the failure with independent, isolated notifications using SAGA's parallel compensation. + - [**Money Transfer**](/core/src/main/java/io/temporal/samples/moneytransfer): Demonstrates the use of a dedicated Activity Worker. - [**Money Batch**](/core/src/main/java/io/temporal/samples/moneybatch): Demonstrates a situation where a single deposit should be initiated for multiple withdrawals. For example, a seller might want to be paid once per fixed number of transactions. This sample can be easily extended to perform a payment based on more complex criteria, such as at a specific time or an accumulated amount. The sample also demonstrates how to Signal the Workflow when it executes (*Signal with start*). If the Workflow is already executing, it just receives the Signal. If it is not executing, then the Workflow executes first, and then the Signal is delivered to it. *Signal with start* is a "lazy" way to execute Workflows when Signaling them. diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflow.java b/core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflow.java new file mode 100644 index 00000000..d0c141fd --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflow.java @@ -0,0 +1,12 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** The critical branch: onboarding cannot succeed without an account. */ +@WorkflowInterface +public interface CreateAccountWorkflow { + + @WorkflowMethod + String createAccount(String userId); +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflowImpl.java b/core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflowImpl.java new file mode 100644 index 00000000..3a5e4ccc --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/CreateAccountWorkflowImpl.java @@ -0,0 +1,18 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.activity.ActivityOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +public class CreateAccountWorkflowImpl implements CreateAccountWorkflow { + + private final OnboardingActivities activities = + Workflow.newActivityStub( + OnboardingActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + @Override + public String createAccount(String userId) { + return activities.createAccount(userId); + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivities.java b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivities.java new file mode 100644 index 00000000..aba8c1eb --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivities.java @@ -0,0 +1,17 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.activity.ActivityInterface; + +/** Activities backing each onboarding branch. */ +@ActivityInterface +public interface OnboardingActivities { + + /** Critical: onboarding cannot succeed without an account. */ + String createAccount(String userId); + + /** Best-effort: storage can be provisioned later out-of-band if this fails. */ + void provisionStorage(String userId); + + /** Best-effort: a missed welcome email is not worth failing onboarding over. */ + void sendWelcomeEmail(String userId); +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivitiesImpl.java b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivitiesImpl.java new file mode 100644 index 00000000..45f1f2e5 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingActivitiesImpl.java @@ -0,0 +1,42 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.failure.ApplicationFailure; + +/** + * Demo implementation. Two sentinel user ids let the sample be run end-to-end against each outcome + * without a test harness: + * + * + */ +public class OnboardingActivitiesImpl implements OnboardingActivities { + + @Override + public String createAccount(String userId) { + if ("user-critical-failure".equals(userId)) { + System.out.println("account creation failing for '" + userId + "'"); + throw ApplicationFailure.newNonRetryableFailure( + "account service rejected the request", "AccountCreationFailure"); + } + System.out.println("created account for '" + userId + "'"); + return "account-" + userId; + } + + @Override + public void provisionStorage(String userId) { + if ("user-degraded".equals(userId)) { + System.out.println("storage provisioning failing for '" + userId + "'"); + throw ApplicationFailure.newNonRetryableFailure( + "storage quota service unavailable", "StorageProvisioningFailure"); + } + System.out.println("provisioned storage for '" + userId + "'"); + } + + @Override + public void sendWelcomeEmail(String userId) { + System.out.println("sent welcome email to '" + userId + "'"); + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingResult.java b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingResult.java new file mode 100644 index 00000000..e58a7cb5 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingResult.java @@ -0,0 +1,48 @@ +package io.temporal.samples.resilientfanout; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Outcome of a successful onboarding run. {@code degradedReasons} is non-empty when one or more + * best-effort branches failed but the run still completed because the critical branch succeeded. + */ +public class OnboardingResult { + + private String accountId; + private List degradedReasons; + + // Default constructor for the JSON payload converter (Workflow results cross the wire). + public OnboardingResult() { + this.degradedReasons = new ArrayList<>(); + } + + public OnboardingResult(String accountId, List degradedReasons) { + this.accountId = accountId; + this.degradedReasons = Collections.unmodifiableList(degradedReasons); + } + + public String getAccountId() { + return accountId; + } + + public List getDegradedReasons() { + return degradedReasons; + } + + @JsonIgnore + public boolean isDegraded() { + return !degradedReasons.isEmpty(); + } + + @Override + public String toString() { + return "OnboardingResult{accountId='" + + accountId + + "', degradedReasons=" + + degradedReasons + + '}'; + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflow.java b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflow.java new file mode 100644 index 00000000..3e620c58 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflow.java @@ -0,0 +1,11 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface OnboardingWorkflow { + + @WorkflowMethod + OnboardingResult onboard(String userId); +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflowImpl.java b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflowImpl.java new file mode 100644 index 00000000..f98b34f2 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/OnboardingWorkflowImpl.java @@ -0,0 +1,112 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.activity.ActivityOptions; +import io.temporal.failure.ChildWorkflowFailure; +import io.temporal.workflow.Async; +import io.temporal.workflow.CancellationScope; +import io.temporal.workflow.ChildWorkflowOptions; +import io.temporal.workflow.Promise; +import io.temporal.workflow.Saga; +import io.temporal.workflow.Workflow; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; + +/** + * Demonstrates a heterogeneous fan-out with partial-failure tolerance: one critical branch and + * several best-effort branches, run as Child Workflows in parallel. + * + *

This is different from a homogeneous fan-out (see the Fan-Out pattern in the Temporal docs), + * where every child does the same kind of work and the run is all-or-nothing. Here the branches do + * different things and have different failure semantics: + * + *

    + *
  • If the critical branch ({@link CreateAccountWorkflow}) fails, the whole run fails, and the + * still-running best-effort branches are cancelled instead of being left to finish and later + * fail trying to report results to an already-dead parent. + *
  • If a best-effort branch fails on its own, the run still succeeds, just in a degraded state + * that callers can inspect via {@link OnboardingResult#getDegradedReasons()}. + *
+ */ +public class OnboardingWorkflowImpl implements OnboardingWorkflow { + + @Override + public OnboardingResult onboard(String userId) { + String parentId = Workflow.getInfo().getWorkflowId(); + + CreateAccountWorkflow createAccountWorkflow = + Workflow.newChildWorkflowStub( + CreateAccountWorkflow.class, childOptions(parentId + "/create-account")); + Promise accountPromise = Async.function(createAccountWorkflow::createAccount, userId); + + // Started inside a CancellationScope so a critical-branch failure can tear them down. + List> bestEffortResults = new ArrayList<>(); + List bestEffortNames = new ArrayList<>(); + CancellationScope bestEffortScope = + Workflow.newCancellationScope( + () -> { + ProvisionStorageWorkflow storageWorkflow = + Workflow.newChildWorkflowStub( + ProvisionStorageWorkflow.class, + childOptions(parentId + "/provision-storage")); + bestEffortNames.add("provision-storage"); + bestEffortResults.add(Async.procedure(storageWorkflow::provisionStorage, userId)); + + SendWelcomeEmailWorkflow emailWorkflow = + Workflow.newChildWorkflowStub( + SendWelcomeEmailWorkflow.class, + childOptions(parentId + "/send-welcome-email")); + bestEffortNames.add("send-welcome-email"); + bestEffortResults.add(Async.procedure(emailWorkflow::sendWelcomeEmail, userId)); + }); + bestEffortScope.run(); + + String accountId; + try { + // Awaited on its own, not via Promise.allOf(...), so it can't be failed by a sibling. + accountId = accountPromise.get(); + } catch (ChildWorkflowFailure e) { + // Cancel explicitly rather than relying on the default ParentClosePolicy (TERMINATE). + bestEffortScope.cancel(); + reconcileFailure(userId, e.getMessage()); + throw e; + } + + // A best-effort failure here degrades the result instead of failing the run. + List degradedReasons = new ArrayList<>(); + for (int i = 0; i < bestEffortResults.size(); i++) { + try { + bestEffortResults.get(i).get(); + } catch (ChildWorkflowFailure e) { + degradedReasons.add(bestEffortNames.get(i) + ": " + e.getMessage()); + } + } + return new OnboardingResult(accountId, degradedReasons); + } + + /** + * Independent cleanup for a failed run: notify support, record an audit entry, and release the + * billing hold. These three calls hit unrelated systems and must not block each other -- one + * failing should not prevent the other two from being attempted. {@code Saga} with parallel + * compensation gives us that isolation for free, even though this isn't classic Saga "undo what I + * did" semantics: there's nothing to undo, only independent notifications to fire. + */ + private void reconcileFailure(String userId, String reason) { + ReconciliationActivities activities = + Workflow.newActivityStub( + ReconciliationActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + Saga saga = new Saga(new Saga.Options.Builder().setParallelCompensation(true).build()); + saga.addCompensation(activities::notifySupportTeam, userId, reason); + saga.addCompensation(activities::recordFailedOnboarding, userId, reason); + saga.addCompensation(activities::releaseBillingHold, userId); + + // Detached so reconciliation still completes while this workflow is unwinding. + Workflow.newDetachedCancellationScope(saga::compensate).run(); + } + + private ChildWorkflowOptions childOptions(String workflowId) { + return ChildWorkflowOptions.newBuilder().setWorkflowId(workflowId).build(); + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflow.java b/core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflow.java new file mode 100644 index 00000000..3021e6aa --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflow.java @@ -0,0 +1,12 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** A best-effort branch: onboarding can still succeed, in a degraded state, without this. */ +@WorkflowInterface +public interface ProvisionStorageWorkflow { + + @WorkflowMethod + void provisionStorage(String userId); +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflowImpl.java b/core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflowImpl.java new file mode 100644 index 00000000..7ced0844 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/ProvisionStorageWorkflowImpl.java @@ -0,0 +1,18 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.activity.ActivityOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +public class ProvisionStorageWorkflowImpl implements ProvisionStorageWorkflow { + + private final OnboardingActivities activities = + Workflow.newActivityStub( + OnboardingActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + @Override + public void provisionStorage(String userId) { + activities.provisionStorage(userId); + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/README.md b/core/src/main/java/io/temporal/samples/resilientfanout/README.md new file mode 100644 index 00000000..4209aed2 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/README.md @@ -0,0 +1,45 @@ +## Resilient fan-out: partial failure tolerance in a heterogeneous DAG + +A user-onboarding Workflow starts three Child Workflows in parallel: + +- `CreateAccountWorkflow` -- **critical**. Onboarding cannot succeed without it. +- `ProvisionStorageWorkflow` -- **best-effort**. Can be redone later if it fails. +- `SendWelcomeEmailWorkflow` -- **best-effort**. Not worth failing onboarding over. + +This is different from a homogeneous fan-out, where every child does the same kind of work and +the parent waits for all of them with `Promise.allOf(...).get()` -- fine when the run really is +all-or-nothing, but wrong here: that call fails fast on whichever promise fails *first*, even if +it's a best-effort branch, and it never tears down the children still running. + +This sample shows three things missing from the existing Child Workflow docs/samples: + +1. **Await each branch individually**, not with `Promise.allOf(...).get()`, so a best-effort + failure doesn't take down a healthy critical branch (or vice versa). +2. **Explicitly cancel the still-running best-effort children** (via a `CancellationScope`) + when the critical branch fails, rather than relying on the default `ParentClosePolicy` + (`TERMINATE`). This isn't just tidiness. `TERMINATE` kills a child outright the moment the + parent closes -- it never delivers a `CanceledFailure`, so the child's own code gets no chance + to react. That matters most when the child has already dispatched real work to a downstream + system before the parent fails: that work doesn't know or care that its Workflow wrapper just + died, so it keeps running on its own, and when it eventually finishes it has no workflow left + to report back to (`WorkflowNotFoundException`). An explicit `.cancel()` at least gives the + child a `CanceledFailure` it can act on before that happens. +3. **Reconcile the failure with independent, isolated side effects.** `Saga` with + `setParallelCompensation(true)` is normally framed as "undo what I already did," but the same + isolation is exactly what you want for a failure-notification step that hits several unrelated + systems, where one call failing must not block the others. + +Run the sample against each outcome: + +```bash +# Happy path -- account created, storage provisioned, email sent. +./gradlew -q execute -PmainClass=io.temporal.samples.resilientfanout.Starter --args="user-1" + +# Best-effort branch fails -- onboarding still succeeds, but degraded. +./gradlew -q execute -PmainClass=io.temporal.samples.resilientfanout.Starter --args="user-degraded" + +# Critical branch fails -- onboarding fails, best-effort siblings are cancelled, failure is reconciled. +./gradlew -q execute -PmainClass=io.temporal.samples.resilientfanout.Starter --args="user-critical-failure" +``` + +Sample unit testing: [OnboardingWorkflowTest](https://github.com/temporalio/samples-java/blob/main/core/src/test/java/io/temporal/samples/resilientfanout/OnboardingWorkflowTest.java) diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivities.java b/core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivities.java new file mode 100644 index 00000000..863a8ba0 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivities.java @@ -0,0 +1,19 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.activity.ActivityInterface; + +/** + * Independent, unrelated side effects run when onboarding fails. Each one talks to a different + * downstream system. They are unrelated to each other on purpose: one failing must not prevent the + * others from being attempted. See {@link OnboardingWorkflowImpl#reconcileFailure} for how this is + * achieved with {@code Saga}. + */ +@ActivityInterface +public interface ReconciliationActivities { + + void notifySupportTeam(String userId, String reason); + + void recordFailedOnboarding(String userId, String reason); + + void releaseBillingHold(String userId); +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivitiesImpl.java b/core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivitiesImpl.java new file mode 100644 index 00000000..0acc64c9 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/ReconciliationActivitiesImpl.java @@ -0,0 +1,19 @@ +package io.temporal.samples.resilientfanout; + +public class ReconciliationActivitiesImpl implements ReconciliationActivities { + + @Override + public void notifySupportTeam(String userId, String reason) { + System.out.println("notifying support team about '" + userId + "': " + reason); + } + + @Override + public void recordFailedOnboarding(String userId, String reason) { + System.out.println("recording failed onboarding audit entry for '" + userId + "': " + reason); + } + + @Override + public void releaseBillingHold(String userId) { + System.out.println("releasing billing hold for '" + userId + "'"); + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflow.java b/core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflow.java new file mode 100644 index 00000000..2df88701 --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflow.java @@ -0,0 +1,12 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +/** A best-effort branch: onboarding can still succeed, in a degraded state, without this. */ +@WorkflowInterface +public interface SendWelcomeEmailWorkflow { + + @WorkflowMethod + void sendWelcomeEmail(String userId); +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflowImpl.java b/core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflowImpl.java new file mode 100644 index 00000000..7943060d --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/SendWelcomeEmailWorkflowImpl.java @@ -0,0 +1,18 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.activity.ActivityOptions; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +public class SendWelcomeEmailWorkflowImpl implements SendWelcomeEmailWorkflow { + + private final OnboardingActivities activities = + Workflow.newActivityStub( + OnboardingActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build()); + + @Override + public void sendWelcomeEmail(String userId) { + activities.sendWelcomeEmail(userId); + } +} diff --git a/core/src/main/java/io/temporal/samples/resilientfanout/Starter.java b/core/src/main/java/io/temporal/samples/resilientfanout/Starter.java new file mode 100644 index 00000000..bf78091f --- /dev/null +++ b/core/src/main/java/io/temporal/samples/resilientfanout/Starter.java @@ -0,0 +1,58 @@ +package io.temporal.samples.resilientfanout; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowOptions; +import io.temporal.envconfig.ClientConfigProfile; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import java.io.IOException; + +public class Starter { + + public static final String TASK_QUEUE = "resilientFanoutTaskQueue"; + + public static void main(String[] args) { + // args[0]: "user-critical-failure", "user-degraded", or anything else (happy path). + String userId = args.length > 0 ? args[0] : "user-1"; + + ClientConfigProfile profile; + try { + profile = ClientConfigProfile.load(); + } catch (IOException e) { + throw new RuntimeException("Failed to load client configuration", e); + } + + WorkflowServiceStubs service = + WorkflowServiceStubs.newServiceStubs(profile.toWorkflowServiceStubsOptions()); + WorkflowClient client = WorkflowClient.newInstance(service, profile.toWorkflowClientOptions()); + WorkerFactory factory = WorkerFactory.newInstance(client); + + Worker worker = factory.newWorker(TASK_QUEUE); + worker.registerWorkflowImplementationTypes( + OnboardingWorkflowImpl.class, + CreateAccountWorkflowImpl.class, + ProvisionStorageWorkflowImpl.class, + SendWelcomeEmailWorkflowImpl.class); + worker.registerActivitiesImplementations( + new OnboardingActivitiesImpl(), new ReconciliationActivitiesImpl()); + factory.start(); + + OnboardingWorkflow workflow = + client.newWorkflowStub( + OnboardingWorkflow.class, + WorkflowOptions.newBuilder() + .setWorkflowId("onboarding-" + userId) + .setTaskQueue(TASK_QUEUE) + .build()); + + try { + OnboardingResult result = workflow.onboard(userId); + System.out.println("Onboarding succeeded: " + result); + } catch (Exception e) { + System.out.println("Onboarding failed: " + e.getMessage()); + } + + System.exit(0); + } +} diff --git a/core/src/test/java/io/temporal/samples/resilientfanout/OnboardingWorkflowTest.java b/core/src/test/java/io/temporal/samples/resilientfanout/OnboardingWorkflowTest.java new file mode 100644 index 00000000..4cb38fec --- /dev/null +++ b/core/src/test/java/io/temporal/samples/resilientfanout/OnboardingWorkflowTest.java @@ -0,0 +1,96 @@ +package io.temporal.samples.resilientfanout; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.temporal.client.WorkflowException; +import io.temporal.failure.ApplicationFailure; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.testing.TestWorkflowExtension; +import io.temporal.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +public class OnboardingWorkflowTest { + + @RegisterExtension + public static final TestWorkflowExtension testWorkflowExtension = + TestWorkflowExtension.newBuilder() + .setWorkflowTypes( + OnboardingWorkflowImpl.class, + CreateAccountWorkflowImpl.class, + ProvisionStorageWorkflowImpl.class, + SendWelcomeEmailWorkflowImpl.class) + .setDoNotStart(true) + .build(); + + @Test + public void testHappyPath( + TestWorkflowEnvironment testEnv, Worker worker, OnboardingWorkflow workflow) { + OnboardingActivities activities = mock(OnboardingActivities.class); + when(activities.createAccount("user-1")).thenReturn("account-user-1"); + ReconciliationActivities reconciliation = mock(ReconciliationActivities.class); + worker.registerActivitiesImplementations(activities, reconciliation); + testEnv.start(); + + OnboardingResult result = workflow.onboard("user-1"); + + assertEquals("account-user-1", result.getAccountId()); + assertFalse(result.isDegraded()); + verify(reconciliation, never()).notifySupportTeam(any(), any()); + } + + @Test + public void testBestEffortFailureDegradesButSucceeds( + TestWorkflowEnvironment testEnv, Worker worker, OnboardingWorkflow workflow) { + OnboardingActivities activities = mock(OnboardingActivities.class); + when(activities.createAccount("user-degraded")).thenReturn("account-user-degraded"); + doThrow( + ApplicationFailure.newNonRetryableFailure( + "storage quota service unavailable", "StorageProvisioningFailure")) + .when(activities) + .provisionStorage("user-degraded"); + ReconciliationActivities reconciliation = mock(ReconciliationActivities.class); + worker.registerActivitiesImplementations(activities, reconciliation); + testEnv.start(); + + OnboardingResult result = workflow.onboard("user-degraded"); + + assertEquals("account-user-degraded", result.getAccountId()); + assertTrue(result.isDegraded()); + assertTrue(result.getDegradedReasons().get(0).startsWith("provision-storage:")); + // The critical branch succeeded, so this was never a failure -- reconciliation must not run. + verify(reconciliation, never()).notifySupportTeam(any(), any()); + } + + @Test + public void testCriticalFailureCancelsSiblingsAndReconciles( + TestWorkflowEnvironment testEnv, Worker worker, OnboardingWorkflow workflow) { + OnboardingActivities activities = mock(OnboardingActivities.class); + when(activities.createAccount("user-critical-failure")) + .thenThrow( + ApplicationFailure.newNonRetryableFailure( + "account service rejected the request", "AccountCreationFailure")); + ReconciliationActivities reconciliation = mock(ReconciliationActivities.class); + worker.registerActivitiesImplementations(activities, reconciliation); + testEnv.start(); + + assertThrows(WorkflowException.class, () -> workflow.onboard("user-critical-failure")); + + // All three reconciliation calls are independent of each other -- verify each was attempted + // exactly once, which is the behavior Saga's parallel compensation gives us here. + verify(reconciliation, times(1)).notifySupportTeam(eq("user-critical-failure"), any()); + verify(reconciliation, times(1)).recordFailedOnboarding(eq("user-critical-failure"), any()); + verify(reconciliation, times(1)).releaseBillingHold(eq("user-critical-failure")); + } +}