Throw CompositeTaskFailedException for context.allOf (#86)

* Throw CompositeTaskFailedException for context.allOf

* Updated ctx.allOf() to throw CompositeTaskFailedException

* Added detailed exception

* Merge changes

* Fixed Java doc

* Catch TaskFailedException from ExecutionException

* Updated exception message

Author: shreyas-gopalakrishna

CHANGELOG.md

@@ -3,6 +3,8 @@
 ### New
 
 * Add CHANGELOG.md file to track changes across versions
+* context.allOf() throws CompositeTaskFailedException(RuntimeException) when one or more tasks fail ([#54](https://github.com/microsoft/durabletask-java/issues/54))
+
 
 ### Updates
 

client/src/main/java/com/microsoft/durabletask/CompositeTaskFailedException.java

@@ -0,0 +1,56 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.microsoft.durabletask;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Exception that gets thrown when multiple {@link Task}s for an activity or sub-orchestration fails with an
+ * unhandled exception.
+ * <p>
+ * Detailed information associated with each task failure can be retrieved using the {@link #getExceptions()}
+ * method.
+ */
+public class CompositeTaskFailedException extends RuntimeException {
+    private final List<Exception> exceptions;
+
+    CompositeTaskFailedException() {
+        this.exceptions = new ArrayList<>();
+    }
+
+    CompositeTaskFailedException(List<Exception> exceptions) {
+        this.exceptions = exceptions;
+    }
+
+    CompositeTaskFailedException(String message, List<Exception> exceptions) {
+        super(message);
+        this.exceptions = exceptions;
+    }
+
+    CompositeTaskFailedException(String message, Throwable cause, List<Exception> exceptions) {
+        super(message, cause);
+        this.exceptions = exceptions;
+    }
+
+    CompositeTaskFailedException(Throwable cause, List<Exception> exceptions) {
+        super(cause);
+        this.exceptions = exceptions;
+    }
+
+    CompositeTaskFailedException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace, List<Exception> exceptions) {
+        super(message, cause, enableSuppression, writableStackTrace);
+        this.exceptions = exceptions;
+    }
+
+    /**
+     * Gets a list of exceptions that occurred during execution of a group of {@link Task}
+     * These exceptions include details of the task failure and exception information
+     *
+     * @return a list of exceptions
+     */
+    public List<Exception> getExceptions() {
+        return new ArrayList<>(this.exceptions);
+    }
+
+}

client/src/main/java/com/microsoft/durabletask/TaskOrchestrationContext.java

@@ -61,7 +61,7 @@ public interface TaskOrchestrationContext {
     //       https://github.com/microsoft/durabletask-java/issues/54
     /**
      * Returns a new {@code Task} that is completed when all the given {@code Task}s complete. If any of the given
-     * {@code Task}s complete with an exception, the returned {@code Task} will also complete with an exception
+     * {@code Task}s complete with an exception, the returned {@code Task} will also complete with an {@link CompositeTaskFailedException}
      * containing details of the first encountered failure. The value of the returned {@code Task} is an ordered list of
      * the return values of the given tasks. If no tasks are provided, returns a {@code Task} completed with value
      * {@code null}.
@@ -76,6 +76,16 @@ public interface TaskOrchestrationContext {
      * List<String> orderedResults = ctx.allOf(List.of(t1, t2, t3)).await();
      * }</pre>
      *
+     * Exceptions in any of the given tasks results in an unchecked {@link CompositeTaskFailedException}.
+     * This exception can be inspected to obtain failure details of individual {@link Task}s.
+     * <pre>{@code
+     * try {
+     *     List<String> orderedResults = ctx.allOf(List.of(t1, t2, t3)).await();
+     * } catch (CompositeTaskFailedException e) {
+     *     List<Exception> exceptions = e.getExceptions()
+     * }
+     * }</pre>
+     *
      * @param tasks the list of {@code Task} objects
      * @param <V> the return type of the {@code Task} objects
      * @return the values of the completed {@code Task} objects in the same order as the source list

client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java

@@ -14,8 +14,6 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
-import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.function.IntFunction;
 import java.util.logging.Logger;
 
@@ -173,23 +171,39 @@ public <V> Task<List<V>> allOf(List<Task<V>> tasks) {
                     .map(t -> t.future)
                     .toArray((IntFunction<CompletableFuture<V>[]>) CompletableFuture[]::new);
 
-            return new CompletableTask<>(CompletableFuture.allOf(futures).thenApply(x -> {
-                ArrayList<V> results = new ArrayList<>(futures.length);
-
-                // All futures are expected to be completed at this point
-                for (CompletableFuture<V> cf : futures) {
-                    try {
-                        results.add(cf.get());
-                    } catch (Exception ex) {
-                        // TODO: Better exception message than this
-                        // TODO: This needs to be a TaskFailedException or some other documented exception type.
-                        //       https://github.com/microsoft/durabletask-java/issues/54
-                        throw new RuntimeException("One or more tasks failed.", ex);
-                    }
-                }
-
-                return results;
-            }));
+            return new CompletableTask<>(CompletableFuture.allOf(futures)
+                    .thenApply(x -> {
+                        List<V> results = new ArrayList<>(futures.length);
+
+                        // All futures are expected to be completed at this point
+                        for (CompletableFuture<V> cf : futures) {
+                            try {
+                                results.add(cf.get());
+                            } catch (Exception ex) {
+                                results.add(null);
+                            }
+                        }
+                        return results;
+                    })
+                    .exceptionally(throwable -> {
+                        ArrayList<Exception> exceptions = new ArrayList<>(futures.length);
+                        for (CompletableFuture<V> cf : futures) {
+                            try {
+                                cf.get();
+                            } catch (ExecutionException ex) {
+                                exceptions.add((Exception) ex.getCause());
+                            } catch (Exception ex){
+                                exceptions.add(ex);
+                            }
+                        }
+                        throw new CompositeTaskFailedException(
+                                String.format(
+                                        "%d out of %d tasks failed with an exception. See the exceptions list for details.",
+                                        exceptions.size(),
+                                        futures.length),
+                                exceptions);
+                    })
+            );
         }
 
         @Override
@@ -1037,6 +1051,10 @@ protected void handleException(Throwable e) throws TaskFailedException {
                     throw (TaskFailedException)e;
                 }
 
+                if (e instanceof CompositeTaskFailedException) {
+                    throw (CompositeTaskFailedException)e;
+                }
+
                 throw new RuntimeException("Unexpected failure in the task execution", e);
             }
 

client/src/test/java/com/microsoft/durabletask/IntegrationTests.java

@@ -8,6 +8,7 @@
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -840,4 +841,61 @@ void waitForInstanceCompletionThrowsException() {
             assertThrows(TimeoutException.class, () -> client.waitForInstanceCompletion(instanceId, Duration.ofSeconds(2), false));
         }
     }
+
+    @Test
+    void activityFanOutWithException() throws TimeoutException {
+        final String orchestratorName = "ActivityFanOut";
+        final String activityName = "Divide";
+        final int count = 10;
+        final String exceptionMessage = "2 out of 6 tasks failed with an exception. See the exceptions list for details.";
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    // Schedule each task to run in parallel
+                    List<Task<Integer>> parallelTasks = IntStream.of(1,2,0,4,0,6)
+                            .mapToObj(i -> ctx.callActivity(activityName, i, Integer.class))
+                            .collect(Collectors.toList());
+
+                    // Wait for all tasks to complete
+                    try {
+                        List<Integer> results = ctx.allOf(parallelTasks).await();
+                        ctx.complete(results);
+                    }catch (CompositeTaskFailedException e){
+                        assertNotNull(e);
+                        assertEquals(2, e.getExceptions().size());
+                        assertEquals(TaskFailedException.class, e.getExceptions().get(0).getClass());
+                        assertEquals(TaskFailedException.class, e.getExceptions().get(1).getClass());
+                        // taskId in the exception below is based on parallelTasks input
+                        assertEquals(getExceptionMessage(activityName, 2, "/ by zero"), e.getExceptions().get(0).getMessage());
+                        assertEquals(getExceptionMessage(activityName, 4, "/ by zero"), e.getExceptions().get(1).getMessage());
+                        throw e;
+                    }
+                })
+                .addActivity(activityName, ctx -> count / ctx.getInput(Integer.class))
+                .buildAndStart();
+
+        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, 0);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.FAILED, instance.getRuntimeStatus());
+
+            List<?> output = instance.readOutputAs(List.class);
+            assertNull(output);
+
+            FailureDetails details = instance.getFailureDetails();
+            assertNotNull(details);
+            assertEquals(exceptionMessage, details.getErrorMessage());
+            assertEquals("com.microsoft.durabletask.CompositeTaskFailedException", details.getErrorType());
+            assertNotNull(details.getStackTrace());
+        }
+    }
+    private static String getExceptionMessage(String taskName, int expectedTaskId, String expectedExceptionMessage) {
+        return String.format(
+                "Task '%s' (#%d) failed with an unhandled exception: %s",
+                taskName,
+                expectedTaskId,
+                expectedExceptionMessage);
+    }
 }