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"));
+ }
+}