Add Resilient Fan-Out sample: partial-failure tolerance in a heterogeneous DAG

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.

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

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

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

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:
+ *
+ * <ul>
+ *   <li>{@code "user-critical-failure"} fails account creation (the critical branch).
+ *   <li>{@code "user-degraded"} fails storage provisioning (a best-effort branch).
+ *   <li>Any other id succeeds on every branch.
+ * </ul>
+ */
+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 + "'");
+  }
+}

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<String> degradedReasons;
+
+  // Default constructor for the JSON payload converter (Workflow results cross the wire).
+  public OnboardingResult() {
+    this.degradedReasons = new ArrayList<>();
+  }
+
+  public OnboardingResult(String accountId, List<String> degradedReasons) {
+    this.accountId = accountId;
+    this.degradedReasons = Collections.unmodifiableList(degradedReasons);
+  }
+
+  public String getAccountId() {
+    return accountId;
+  }
+
+  public List<String> getDegradedReasons() {
+    return degradedReasons;
+  }
+
+  @JsonIgnore
+  public boolean isDegraded() {
+    return !degradedReasons.isEmpty();
+  }
+
+  @Override
+  public String toString() {
+    return "OnboardingResult{accountId='"
+        + accountId
+        + "', degradedReasons="
+        + degradedReasons
+        + '}';
+  }
+}

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

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.
+ *
+ * <p>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:
+ *
+ * <ul>
+ *   <li>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.
+ *   <li>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()}.
+ * </ul>
+ */
+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<String> accountPromise = Async.function(createAccountWorkflow::createAccount, userId);
+
+    // Started inside a CancellationScope so a critical-branch failure can tear them down.
+    List<Promise<Void>> bestEffortResults = new ArrayList<>();
+    List<String> 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<String> 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();
+  }
+}

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

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

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)

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

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