Replacing System.AggregateException with custom aggregate exception

src/lib/Microsoft.Health.Common/Service/ParallelTaskWorker.cs

@@ -66,7 +66,7 @@ await consumer.Completion
                     {
                         if (!exceptions.IsEmpty)
                         {
-                            throw new AggregateException(exceptions);
+                            throw new SimpleAggregateException(exceptions);
                         }
                     },
                     cts.Token,

src/lib/Microsoft.Health.Common/SimpleAggregateException.cs

@@ -0,0 +1,54 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using System.Text;
+using EnsureThat;
+
+namespace Microsoft.Health.Common
+{
+    public class SimpleAggregateException : Exception
+    {
+        private readonly string _aggregatedMessage;
+
+        public SimpleAggregateException(IEnumerable<Exception> exceptions)
+            : this(new List<Exception>(exceptions))
+        {
+        }
+
+        public SimpleAggregateException(ICollection<Exception> exceptions)
+            : this("One or more exceptions have occured", exceptions)
+        {
+        }
+
+        public SimpleAggregateException(string message, ICollection<Exception> exceptions)
+        {
+            EnsureArg.IsNotNullOrWhiteSpace(message, nameof(message));
+            EnsureArg.HasItems(exceptions, nameof(exceptions));
+
+            _aggregatedMessage = ConstructMessage(message, exceptions);
+            InnerExceptions = new List<Exception>(exceptions);
+        }
+
+        public override string Message => _aggregatedMessage;
+
+        public IReadOnlyCollection<Exception> InnerExceptions { get; }
+
+        private string ConstructMessage(string message, IEnumerable<Exception> exceptions)
+        {
+            var exceptionMessage = new StringBuilder(message);
+
+            exceptionMessage.AppendLine(":");
+
+            foreach (var exception in exceptions)
+            {
+                exceptionMessage.AppendLine($"---- {exception.GetType()}: {exception.Message}");
+            }
+
+            return exceptionMessage.ToString();
+        }
+    }
+}

src/lib/Microsoft.Health.Fhir.Ingest/Service/MeasurementEventNormalizationService.cs

@@ -12,6 +12,7 @@
 using EnsureThat;
 using Microsoft.Azure.EventHubs;
 using Microsoft.Azure.WebJobs;
+using Microsoft.Health.Common;
 using Microsoft.Health.Fhir.Ingest.Data;
 using Microsoft.Health.Fhir.Ingest.Telemetry;
 using Microsoft.Health.Fhir.Ingest.Template;
@@ -124,7 +125,7 @@ await consumer.Completion
                     {
                         if (!exceptions.IsEmpty)
                         {
-                            throw new AggregateException(exceptions);
+                            throw new SimpleAggregateException(exceptions);
                         }
                     },
                     cts.Token,

test/Microsoft.Health.Common.UnitTests/SimpleAggregateExceptionTest.cs

@@ -0,0 +1,50 @@
+// -------------------------------------------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License (MIT). See LICENSE in the repo root for license information.
+// -------------------------------------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Xunit;
+
+namespace Microsoft.Health.Common.UnitTests
+{
+    public class SimpleAggregateExceptionTest
+    {
+        [Fact]
+        public void When_BadArgumentsSupplied_ExceptionIsThrown()
+        {
+            Assert.Throws<ArgumentException>(() => new SimpleAggregateException(new List<Exception>()));
+            Assert.Throws<ArgumentException>(() => new SimpleAggregateException(string.Empty, new List<Exception>() { new Exception() }));
+        }
+
+        [Fact]
+        public void When_SuppliedWithExceptions_ExpectedMessageCreated()
+        {
+            var exceptions = Enumerable.Range(0, 3).Select(i => new Exception("Mock Exception")).ToList();
+            var expectedMessage = @"One or more exceptions have occured:
+---- System.Exception: Mock Exception
+---- System.Exception: Mock Exception
+---- System.Exception: Mock Exception
+";
+            var exception = new SimpleAggregateException(exceptions);
+            Assert.Equal(3, exception.InnerExceptions.Count);
+            Assert.Equal(expectedMessage, exception.Message);
+        }
+
+        [Fact]
+        public void When_SuppliedWithExceptions_AndMessage_ExpectedMessageCreated()
+        {
+            var exceptions = Enumerable.Range(0, 3).Select(i => new Exception("Mock Exception")).ToList();
+            var expectedMessage = @"Custom exception message:
+---- System.Exception: Mock Exception
+---- System.Exception: Mock Exception
+---- System.Exception: Mock Exception
+";
+            var exception = new SimpleAggregateException("Custom exception message", exceptions);
+            Assert.Equal(3, exception.InnerExceptions.Count);
+            Assert.Equal(expectedMessage, exception.Message);
+        }
+    }
+}

test/Microsoft.Health.Fhir.Ingest.UnitTests/Service/MeasurementEventNormalizationServiceTests.cs

@@ -8,6 +8,7 @@
 using System.Threading.Tasks;
 using Microsoft.Azure.EventHubs;
 using Microsoft.Azure.WebJobs;
+using Microsoft.Health.Common;
 using Microsoft.Health.Fhir.Ingest.Data;
 using Microsoft.Health.Fhir.Ingest.Template;
 using Microsoft.Health.Logging.Telemetry;
@@ -106,6 +107,30 @@ await consumer.Received(1)
             }
         }
 
+        [Fact]
+        public async Task GivenEvents_WhenProcessAsync_AndGetMeasurementThrowsException_ThenNoEventsAreConsumed_Test()
+        {
+            var template = Substitute.For<IContentTemplate>();
+
+            template.GetMeasurements(null).ReturnsForAnyArgs(arg => throw new Exception("Mock Exception"));
+
+            var converter = Substitute.For<Data.IConverter<EventData, JToken>>();
+
+            var events = Enumerable.Range(0, 10).Select(i => BuildEvent(i)).ToArray();
+
+            var log = Substitute.For<ITelemetryLogger>();
+
+            var consumer = Substitute.For<IAsyncCollector<IMeasurement>>();
+
+            var srv = new MeasurementEventNormalizationService(log, template, converter, 1);
+            var exception = await Assert.ThrowsAsync<SimpleAggregateException>(() => srv.ProcessAsync(events, consumer));
+
+            Assert.Equal(events.Length, exception.InnerExceptions.Count);
+            template.ReceivedWithAnyArgs(10).GetMeasurements(null);
+            converter.ReceivedWithAnyArgs(10).Convert(null);
+            await consumer.ReceivedWithAnyArgs(0).AddAsync(null);
+        }
+
         [Fact]
         public async Task GivenEventsAndDefaultErrorConsumer_WhenProcessAsyncAndConsumerErrors_ThenEachEventResultConsumedAndErrorProprogated_Test()
         {
@@ -121,7 +146,7 @@ public async Task GivenEventsAndDefaultErrorConsumer_WhenProcessAsyncAndConsumer
             consumer.AddAsync(null).ReturnsForAnyArgs(v => Task.FromException(new Exception()));
 
             var srv = new MeasurementEventNormalizationService(log, template, converter, 1);
-            var exception = await Assert.ThrowsAsync<AggregateException>(() => srv.ProcessAsync(events, consumer));
+            var exception = await Assert.ThrowsAsync<SimpleAggregateException>(() => srv.ProcessAsync(events, consumer));
             Assert.Equal(events.Length, exception.InnerExceptions.Count);
 
             template.ReceivedWithAnyArgs(events.Length).GetMeasurements(null);

test/Microsoft.Health.Fhir.Ingest.UnitTests/Service/MeasurementFhirImportServiceTests.cs

@@ -8,6 +8,7 @@
 using System.IO;
 using System.Linq;
 using System.Threading.Tasks;
+using Microsoft.Health.Common;
 using Microsoft.Health.Common.Config;
 using Microsoft.Health.Common.Telemetry;
 using Microsoft.Health.Fhir.Ingest.Config;
@@ -76,7 +77,7 @@ public async void GivenExceptionDuringParseStreamAsync_WhenParseStreamAsync_Then
 
             var measurements = new MeasurementGroup[] { Substitute.For<MeasurementGroup>(), Substitute.For<MeasurementGroup>() };
 
-            var aggEx = await Assert.ThrowsAsync<AggregateException>(async () => await fhirImport.ProcessStreamAsync(ToStream(measurements), string.Empty, log));
+            var aggEx = await Assert.ThrowsAsync<SimpleAggregateException>(async () => await fhirImport.ProcessStreamAsync(ToStream(measurements), string.Empty, log));
 
             Assert.Collection(
                 aggEx.InnerExceptions,