Fix exception type when use RetriableTask in Fan in/out pattern

[kaibocai]

Issue describing the changes in this PR

Resolve #169

This PR fixes the exception type for using RetriableTask in allOf method.

Description of the issue:

For RetriableTask in allOf method if the task failed for exception, CompositeTaskFailedException is not thrown out instead it's a TaskFailedException being thrown out. The reason is that the CompositeTaskFailedException that built in exceptionPath haven't get chance to be obtained at future.get(), ContextImplTask.this.processNextEvent() will first threw a TaskFailedException

For example:

 @FunctionName("Parallel")
    public List<String> parallelOrchestratorSad(
            @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx,
            ExecutionContext context) {
        try {
            List<Task<String>> tasks = new ArrayList<>();
            RetryPolicy policy = new RetryPolicy(2, Duration.ofSeconds(15));
            TaskOptions options = new TaskOptions(policy);
            tasks.add(ctx.callActivity("AppendSad", "Input1", options, String.class));
            tasks.add(ctx.callActivity("AppendHappy", "Input2", options, String.class));
            return ctx.allOf(tasks).await();
        } catch (CompositeTaskFailedException e) {
            // Nothing will be caught for this type of exception. 
            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());
                }
            }
        }
        return null;
    }
    
    @FunctionName("AppendHappy")
    public String appendHappy(
            @DurableActivityTrigger(name = "name") String name,
            final ExecutionContext context) {
        context.getLogger().info("AppendHappy: " + name);
        return name + "-test-happy";
    }
    
    @FunctionName("AppendSad")
    public String appendSad(
            @DurableActivityTrigger(name = "name") String name,
            final ExecutionContext context) {
        context.getLogger().info("Throw Test Exception: " + name);
        throw new NullPointerException("Test kaibocai exception");
    }

This PR update the do-while block logics so that all exception will be handle inside the block

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

Line 1264 in e938278

} while (ContextImplTask.this.processNextEvent());

Pull request checklist

• My changes do not require documentation changes
  • Otherwise: Documentation issue linked to PR
• My changes are added to the CHANGELOG.md
• I have added all required tests (Unit tests, E2E tests)

Additional information

Additional PR information

[davidmrdavid]

LGTM so long as the suggested comment edit is incorporated.

[davidmrdavid]

just to confirm: can processNextEvent actually lead to a ContinueAsNewInterruption? My understanding is that processNextEvent just iterates through the history, and I would expect that ContinueAsNewInterruption would only occur when the customer code is actually invoked.

[kaibocai]

Yes, ContinueAsNewInterruption would only occur when the customer code is actually invoked. The SDK code will never cause a ContinueAsNewInterruption. But when we iterate the history, if the customer code is throwing ContinueAsNewInterruption, we will let it go at this place instead of catching it.

[davidmrdavid]

some nits on phrasing:

Suggested change

	// ignore
	/**
	* the idea is that for any exception thrown here it should be handled
	* in the do-while loop, the same exception should also be obtained when calling
	* {code#future.get()} in the do-while loop, and it will be handled correctly
	* in the catch block in the do-while loop.
	*/
	}
	// Any exception happen we return true so that we will enter to the do-while block for the last time.
	/**
	* 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.
	*/
	}
	// On an exception, we return true to the do-while block one last time.