I've created an asynchronous parallel ForEach mechanism so I could enumerate an enumerable by N degrees of concurrency and process an action for each item. Additionally, I have the option of stopping when first exception is encountered or processing the entire enumerable and returning all exceptions at the end. I'm trying to ensure that all my exception handling is being done correctly and I understand the flow correctly. My original code review can be found here (Original Answer). The code below incorporates suggestions from the original answer, but I'm not sure all the assumptions stated in that answer are happening.

My assumptions and concerns (some contrary to original answer) are posted with each test case. Please feel free to correct any incorrect assumptions.

Test Harness In All Test Cases

// This really would be a CreateLinkedTokenSource that could be cancelled
// from the caller of this code or internally within the body of the ForEachAsync body
var internalCancel = new CancellationTokenSource();

try
{
    // Simulate running something for 10 data batches...
    await Enumerable.Range(0, 10)
        .ForEachAsync(
            async jobData =>
            {
                // body of code that might throw exceptions or set internalCancel.Cancel()...

                internalCancel.Token.ThrowIfCancellationRequested();

                Console.WriteLine( $"Task {jobKey}: FooAsync - Finish DataChunk {jobData}" );
            },
            new AsyncParallelOptions { MaxDegreeOfParallelism = 4, CancellationToken = internalCancel.Token }
        );
}
catch (Exception ex)
{
    Console.WriteLine( $"Task {jobKey}: FooAsync - Exception: {ex.GetType().ToString()}, internalCancel.Token.IsCancellationRequested: {internalCancel.Token.IsCancellationRequested}" );
    throw;
}

Test Case 1 - Run body() to completion

This seems to work as expected.

Test Case 2 - Inside body() exception is thrown

Inside the body(), throw a NotSupportedException() (or any exception other than OperationCancelledException) and I get the following output.

ForEachAsync Extension #1: ContinueWith, t.Exception is null: False, t.IsCanceled: False, t.IsFaulted: True, cts.IsCancellationRequested: False
ForEachAsync Extension #2: Finished, 1 total, allDone.IsCanceled: False, allDone.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #3: Throw Exceptions
Task 0: FooAsync - Exception: System.AggregateException, internalCancel.Token.IsCancellationRequested: False

My assumptions/concerns are:

1. Any exceptions thrown in body() are swallowed and presented inside ContinueWith() via t.Exception or t.IsFaulted.
2. Since I did not throw t.Exception and did not use cts.Token.ThrowIfCancellationRequested(), that is the reason why after await allDone, allDone.IsFaulted and allDone.IsCanceled is always false?
3. Original answer mentioned allDone.IsFaulted could be true if body() threw synchronously. I'm not sure what throwing synchronously means, but as my code stands I couldn't seem to ever have it set to true.
4. Original answer mentioned code after await allDone; would not run if thrown synchronously, but as my output displays, code is running after the await allDone. Is this because I didn't throw any exceptions from inside ContinueWith()?

Test Case 3 - Inside body() OperationCancelledException is thrown

Output:

ForEachAsync Extension #1: ContinueWith, t.Exception is null: True, t.IsCanceled: True, t.IsFaulted: False, cts.IsCancellationRequested: False
// Next three are dumps from the other 3 degrees of parallelism
ForEachAsync Extension #1: ContinueWith, t.Exception is null: True, t.IsCanceled: False, t.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #1: ContinueWith, t.Exception is null: True, t.IsCanceled: False, t.IsFaulted: False, cts.IsCancellationRequested: False
ForEachAsync Extension #1: ContinueWith, t.Exception is null: True, t.IsCanceled: False, t.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #2: Finished, 0 total, allDone.IsCanceled: False, allDone.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #4: Throw OperationCanceledException
Task 0: FooAsync - Exception: System.OperationCanceledException, internalCancel.Token.IsCancellationRequested: False

Assumptions/Concerns:

1. Original answer stated that it was better practice if I used cts.Token.ThrowIfCancellationRequested() instead of checking cts.IsCancellationRequested however, I assume I'd place the ThrowIfCancellationRequested() as first line in while( partition.MoveNext() ) { } which means that if an exception or cancellation happened during the body, the next partition would be prepared for processing which could involve hitting web services and/or database calls. Checking cts.IsCancellationRequested seems to be a more efficient approach? Unless I'm over thinking things.
2. Original answer stated that I could just re throw t.Exception instead of adding to my collection then I could get my exception handling for free. I assume this means that await allDone; would then throw the exception and no code after that would be executed? This raises two questions.
   1. If I wanted entire enumerable processed and all exceptions thrown at end, this wouldn't allow that right? This would stop execution as soon as the first exception occurred?
   2. The suggestion to call t.Exception.Flatten() before adding to my exception would be lost, then requiring the caller of ForEachAsync to do that instead?

Test Case 4 - Inside body() internalCancel.Token.ThrowIfCancellationRequested() is triggered

Output:

ForEachAsync Extension #1: ContinueWith, t.Exception is null: True, t.IsCanceled: True, t.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #1: ContinueWith, t.Exception is null: True, t.IsCanceled: True, t.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #2: Finished, 0 total, allDone.IsCanceled: False, allDone.IsFaulted: False, cts.IsCancellationRequested: True
ForEachAsync Extension #4: Throw OperationCanceledException
Task 0: FooAsync - Exception: System.OperationCanceledException, internalCancel.Token.IsCancellationRequested: True

Assumptions/Concerns:

1. Similar to the above case, since I don't use cts.Token.ThrowIfCancellationRequested(); my while loop exits without exception. Then, exceptions.Count==0 so I simply throw a new OperationCancelledException and the flow seems to behave correctly (or at least how I expect it to).

If you've made it this far, I appreciate any and all comments.

ForEachAsync Extension Method

    public static async Task ForEachAsync<T>( this IEnumerable<T> source, Func<T, Task> body, AsyncParallelOptions parallelOptions )
    {
        ConcurrentBag<Exception> exceptions = new ConcurrentBag<Exception>();

        var maxDegreeOfConcurrency = parallelOptions.MaxDegreeOfParallelism;

        // If they pass in a CancellationToken from caller of ForEachAsync need to create linked token source in case caller cancels, I want
        // ForEachAsync to cancel as well.  If they want to failImmediately, make a new CancellationTokenSource so I can stop processing partitions
        var cts = parallelOptions.CancellationToken != CancellationToken.None
            ? CancellationTokenSource.CreateLinkedTokenSource( parallelOptions.CancellationToken )
            : new CancellationTokenSource();

        var allDone = Task.WhenAll(
            from partition in Partitioner.Create( source ).GetPartitions( maxDegreeOfConcurrency )
            select Task.Run( async delegate
            {
                using ( partition )
                {
                    while ( !cts.IsCancellationRequested /* either from caller or failImmediately */ && partition.MoveNext() )
                    {
                        await body( partition.Current ).ContinueWith( t =>
                           {
                               Console.WriteLine( $"ForEachAsync Extension #1: ContinueWith, t.Exception is null: {t.Exception == null}, t.IsCanceled: {t.IsCanceled}, t.IsFaulted: {t.IsFaulted}, cts.IsCancellationRequested: {cts.IsCancellationRequested}" );

                               // If body() threw an error, cancel if a CancellationTokenSource is present.
                               if ( t.Exception != null )
                               {
                                   if ( parallelOptions.FailImmediately )
                                   {
                                       cts.Cancel();
                                   }

                                   // Always gather the exception to throw at the end
                                   foreach ( var ex in t.Exception.Flatten().InnerExceptions )
                                   {
                                       exceptions.Add( ex );
                                   }
                               }
                               else if (t.IsCanceled)
                               {
                                    // Needed in case OperationCanceledException() is thrown manually without calling 
                                    // .Cancel() on any linked token sources
                                    cts.Cancel(); 
                               }                                   
                           }
                        );
                    }
                }
            }, cts.Token ) );

        // Wait until all finished (or errored out) and then return exceptions
        await allDone;

        // Question: allDone is never going to have IsCanceled or IsFaulted correct?  because await body() will swallow all exceptions?
        Console.WriteLine( $"ForEachAsync Extension #2: Finished, {exceptions?.Count ?? 0} total, allDone.IsCanceled: {allDone.IsCanceled}, allDone.IsFaulted: {allDone.IsFaulted}, cts.IsCancellationRequested: {cts.IsCancellationRequested}" );

        if ( exceptions.Count > 0 )
        {
            Console.WriteLine( $"ForEachAsync Extension #3: Throw Exceptions" );
            throw new AggregateException( exceptions );
        }

        if ( cts.IsCancellationRequested )
        {
            Console.WriteLine( $"ForEachAsync Extension #4: Throw OperationCanceledException" );
            throw new OperationCanceledException();
        }
    }