Fix exception type when use RetriableTask in Fan in/out pattern (#174)

* fix exception type when use RetriableTask in Fan in/out pattern

* update CHANGELOG

* rethrow OrchestratorBlockedException and ContinueAsNewInterruption

* minor updates

Author: kaibocai

CHANGELOG.md

@@ -1,3 +1,7 @@
+## placeholder
+* Fix exception type issue when using `RetriableTask` in fan in/out pattern ([#174](https://github.com/microsoft/durabletask-java/pull/174))
+
+
 ## v1.4.0
 
 ### Updates

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

@@ -1261,7 +1261,7 @@ public V await() {
                             this.handleException(e);
                         }
                     }
-                } while (ContextImplTask.this.processNextEvent());
+                } while (processNextEvent());
 
                 // There's no more history left to replay and the current task is still not completed. This is normal.
                 // The OrchestratorBlockedException exception allows us to yield the current thread back to the executor so
@@ -1271,6 +1271,22 @@ public V await() {
                         "The orchestrator is blocked and waiting for new inputs. This Throwable should never be caught by user code.");
             }
 
+            private boolean processNextEvent() {
+                try {
+                    return ContextImplTask.this.processNextEvent();
+                } catch (OrchestratorBlockedException | ContinueAsNewInterruption exception) {
+                    throw exception;
+                } catch (Exception e) {
+                    // ignore
+                    /**
+                     * We ignore the exception. Any Durable Task exceptions thrown here can be obtained when calling
+                     * {code#future.get()} in the implementation of 'await'. We defer to that loop to handle the exception.
+                     */
+                }
+                // Any exception happen we return true so that we will enter to the do-while block for the last time.
+                return true;
+            }
+
             @Override
             public <U> CompletableTask<U> thenApply(Function<V, U> fn) {
                 CompletableFuture<U> newFuture = this.future.thenApply(fn);

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

@@ -1374,6 +1374,68 @@ void activityAllOf() throws IOException, TimeoutException {
         }
     }
 
+    @Test
+    void activityAllOfException() throws IOException, TimeoutException {
+        final String orchestratorName = "ActivityAllOf";
+        final String activityName = "ToString";
+        final String retryActivityName = "RetryToStringException";
+        final String result = "test fail";
+        final int activityMiddle = 5;
+        final RetryPolicy retryPolicy = new RetryPolicy(2, Duration.ofSeconds(5));
+        final TaskOptions taskOptions = new TaskOptions(retryPolicy);
+
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> {
+                    List<Task<String>> parallelTasks = IntStream.range(0, activityMiddle * 2)
+                            .mapToObj(i -> {
+                                if (i < activityMiddle) {
+                                    return ctx.callActivity(activityName, i, String.class);
+                                } else {
+                                    return ctx.callActivity(retryActivityName, i, taskOptions, String.class);
+                                }
+                            })
+                            .collect(Collectors.toList());
+
+                    // Wait for all tasks to complete, then sort and reverse the results
+                    try {
+                        List<String> results = null;
+                        results = ctx.allOf(parallelTasks).await();
+                        Collections.sort(results);
+                        Collections.reverse(results);
+                        ctx.complete(results);
+                    } catch (CompositeTaskFailedException e) {
+                        // only catch this type of exception to ensure the expected type of exception is thrown out.
+                        for (Exception exception : e.getExceptions()) {
+                            if (exception instanceof TaskFailedException) {
+                                TaskFailedException taskFailedException = (TaskFailedException) exception;
+                                System.out.println("Task: " + taskFailedException.getTaskName() +
+                                        " Failed for cause: " + taskFailedException.getErrorDetails().getErrorMessage());
+                            }
+                        }
+                    }
+                    ctx.complete(result);
+                })
+                .addActivity(activityName, ctx -> ctx.getInput(Object.class).toString())
+                .addActivity(retryActivityName, ctx -> {
+                    // only throw exception
+                    throw new RuntimeException("test retry");
+                })
+                .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.COMPLETED, instance.getRuntimeStatus());
+
+            String output = instance.readOutputAs(String.class);
+            assertNotNull(output);
+            assertEquals(String.class, output.getClass());
+            assertEquals(result, output);
+        }
+    }
+
     @Test
     void activityAnyOf() throws IOException, TimeoutException {
         final String orchestratorName = "ActivityAnyOf";

samples-azure-functions/src/main/java/com/functions/ParallelFunctions.java

@@ -89,4 +89,49 @@ public Object parallelAnyOf(
         tasks.add(ctx.callActivity("AppendHappy", 1, Integer.class));
         return ctx.anyOf(tasks).await().await();
     }
+
+    @FunctionName("StartParallelCatchException")
+    public HttpResponseMessage startParallelCatchException(
+            @HttpTrigger(name = "req", methods = {HttpMethod.GET, HttpMethod.POST}, authLevel = AuthorizationLevel.ANONYMOUS) HttpRequestMessage<Optional<String>> request,
+            @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
+            final ExecutionContext context) {
+        context.getLogger().info("Java HTTP trigger processed a request.");
+
+        DurableTaskClient client = durableContext.getClient();
+        String instanceId = client.scheduleNewOrchestrationInstance("ParallelCatchException");
+        context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
+        return durableContext.createCheckStatusResponse(request, instanceId);
+    }
+
+    @FunctionName("ParallelCatchException")
+    public List<String> parallelCatchException(
+            @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx,
+            ExecutionContext context) {
+        try {
+            List<Task<String>> tasks = new ArrayList<>();
+            RetryPolicy policy = new RetryPolicy(2, Duration.ofSeconds(1));
+            TaskOptions options = new TaskOptions(policy);
+            tasks.add(ctx.callActivity("AlwaysException", "Input1", options, String.class));
+            tasks.add(ctx.callActivity("AppendHappy", "Input2", options, String.class));
+            return ctx.allOf(tasks).await();
+        } catch (CompositeTaskFailedException e) {
+            // only catch this type of exception to ensure the expected type of exception is thrown out.
+            for (Exception exception : e.getExceptions()) {
+                if (exception instanceof TaskFailedException) {
+                    TaskFailedException taskFailedException = (TaskFailedException) exception;
+                    context.getLogger().info("Task: " + taskFailedException.getTaskName() +
+                            " Failed for cause: " + taskFailedException.getErrorDetails().getErrorMessage());
+                }
+            }
+        }
+        return null;
+    }
+
+    @FunctionName("AlwaysException")
+    public String alwaysException(
+            @DurableActivityTrigger(name = "name") String name,
+            final ExecutionContext context) {
+        context.getLogger().info("Throw Test AlwaysException: " + name);
+        throw new RuntimeException("Test AlwaysException");
+    }
 }

samples-azure-functions/src/test/java/com/functions/EndToEndTests.java

@@ -33,7 +33,8 @@ public void setupHost() {
     @ValueSource(strings = {
             "StartOrchestration",
             "StartParallelOrchestration",
-            "StartParallelAnyOf"
+            "StartParallelAnyOf",
+            "StartParallelCatchException"
     })
     public void generalFunctions(String functionName) throws InterruptedException {
         Set<String> continueStates = new HashSet<>();