Adding detailed failure event handler for DataMovement Tests and fixed Cancellation Exception event being raised unnecessarily (Azure#35441)

* WIP - made failuretransferholder, still need to apply it to the rest of the tests

* Updated single download tests

* Added FailureTransferHolder to rest of tests

* Corrected failure tests to check the error messages instead of asserting failure on expected failures

* Remove assert failure check on expected failure tests; Fixed bug where cancellation exception gets event raised

* Undo reenabling pause and resume tests

* Actually reverting the pause tests to be ignored instead of liveonly

Author: amnguye

sdk/storage/Azure.Storage.DataMovement/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## 12.0.0-beta.2 (Unreleased)
 - Fix to prevent thread starvation on the DataTransfer.AwaitCompletion
+- Fix to prevent unnecessary OperationCancelledException's showing up in the SingleTransferOptions.TransferFailed and ContainerTransferOptions.TransferFailed when cancelling a job.
 
 ## 12.0.0-beta.1 (2022-12-15)
 - This preview is the first release of a ground-up rewrite of our client data movement

sdk/storage/Azure.Storage.DataMovement/src/CommitChunkHandler.cs

@@ -182,10 +182,6 @@ await _invokeFailedEventHandler(
                     _currentBytesSemaphore.Release();
                 }
             }
-            catch (OperationCanceledException)
-            {
-                // If operation cancelled, no need to log the exception. As it's logged by whoever called the cancellation (e.g. disposal)
-            }
             catch (Exception ex)
             {
                 await _invokeFailedEventHandler(ex).ConfigureAwait(false);

sdk/storage/Azure.Storage.DataMovement/src/DownloadChunkHandler.cs

@@ -254,10 +254,6 @@ await InvokeFailedEvent(
                     _currentBytesSemaphore.Release();
                 }
             }
-            catch (OperationCanceledException)
-            {
-                // If operation cancelled, no need to log the exception. As it's logged by whoever called the cancellation (e.g. disposal)
-            }
             catch (Exception ex)
             {
                 await _invokeFailedEventHandler(ex).ConfigureAwait(false);

sdk/storage/Azure.Storage.DataMovement/src/JobPartInternal.cs

@@ -334,7 +334,8 @@ await TransferSkippedEventHandler.Invoke(new TransferSkippedEventArgs(
         /// </summary>
         public async virtual Task InvokeFailedArg(Exception ex)
         {
-            if (TransferFailedEventHandler != null)
+            if (TransferFailedEventHandler != null &&
+                ex is not OperationCanceledException)
             {
                 // TODO: change to RaiseAsync
                 await TransferFailedEventHandler.Invoke(new TransferFailedEventArgs(

sdk/storage/Azure.Storage.DataMovement/src/ServiceToServiceJobPart.cs

@@ -424,13 +424,6 @@ await _commitBlockHandler.InvokeEvent(
                         true,
                         _cancellationToken)).ConfigureAwait(false);
             }
-            // If we fail to stage a block, we need to make sure the rest of the stage blocks are cancelled
-            // (Core already performs the retry policy on the one stage block request
-            // which means the rest are not worth to continue)
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-            }
             catch (RequestFailedException ex)
             when (_createMode == StorageResourceCreateMode.Overwrite
                     && ex.ErrorCode == "BlobAlreadyExists")

sdk/storage/Azure.Storage.DataMovement/src/StreamToUriJobPart.cs

@@ -268,10 +268,6 @@ await _destinationResource.WriteFromStreamAsync(
             {
                 await InvokeSkippedArg().ConfigureAwait(false);
             }
-            catch (OperationCanceledException)
-            {
-                // expected exception during job part cancellation or pause.
-            }
             catch (Exception ex)
             {
                 await InvokeFailedArg(ex).ConfigureAwait(false);
@@ -343,13 +339,6 @@ await _commitBlockHandler.InvokeEvent(
                         true,
                         _cancellationToken)).ConfigureAwait(false);
             }
-            // If we fail to stage a block, we need to make sure the rest of the stage blocks are cancelled
-            // (Core already performs the retry policy on the one stage block request
-            // which means the rest are not worth to continue)
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-            }
             catch (Exception ex)
             {
                 await InvokeFailedArg(ex).ConfigureAwait(false);

sdk/storage/Azure.Storage.DataMovement/src/UriToStreamJobPart.cs

@@ -145,11 +145,6 @@ public override async Task ProcessPartToChunkAsync()
                     await LengthKnownDownloadInternal().ConfigureAwait(false);
                 }
             }
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-                await OnTransferStatusChanged(StorageTransferStatus.None).ConfigureAwait(false);
-            }
             catch (Exception ex)
             {
                 // The file either does not exist any more, got moved, or renamed.
@@ -342,11 +337,6 @@ await _downloadChunkHandler.InvokeEvent(new DownloadRangeEventArgs(
                     false,
                     _cancellationToken)).ConfigureAwait(false);
             }
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-                await OnTransferStatusChanged(StorageTransferStatus.None).ConfigureAwait(false);
-            }
             catch (Exception ex)
             {
                 // Unexpected exception
@@ -382,11 +372,6 @@ await _destinationResource.WriteFromStreamAsync(
                 // Skip file that already exsits on the destination.
                 await InvokeSkippedArg().ConfigureAwait(false);
             }
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-                await OnTransferStatusChanged(StorageTransferStatus.None).ConfigureAwait(false);
-            }
             catch (Exception ex)
             {
                 await InvokeFailedArg(ex).ConfigureAwait(false);
@@ -409,11 +394,6 @@ await source.CopyToAsync(
                         .ConfigureAwait(false);
                 }
             }
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-                await OnTransferStatusChanged(StorageTransferStatus.None).ConfigureAwait(false);
-            }
             catch (Exception ex)
             {
                 await InvokeFailedArg(ex).ConfigureAwait(false);
@@ -430,11 +410,6 @@ private async Task FlushFinalCryptoStreamInternal()
             {
                 await _destinationResource.CompleteTransferAsync(_cancellationToken).ConfigureAwait(false);
             }
-            catch (OperationCanceledException)
-            {
-                // Job was cancelled
-                await OnTransferStatusChanged(StorageTransferStatus.None).ConfigureAwait(false);
-            }
             catch (Exception ex)
             {
                 await InvokeFailedArg(ex).ConfigureAwait(false);

sdk/storage/Azure.Storage.DataMovement/tests/FailureTrackingTransfer.cs

@@ -0,0 +1,80 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Azure.Storage.DataMovement.Models;
+using NUnit.Framework;
+
+namespace Azure.Storage.DataMovement.Tests
+{
+    /// <summary>
+    /// Stores the option bag and failures as the test progresses.
+    ///
+    /// This prevents a test from being torn down terribly when doing
+    /// an Assert.Failure in the middle of an event.
+    ///
+    /// Alos if there's multiple failures then we will catch all of them.
+    /// (Which would mainly occur during <see cref="ErrorHandlingOptions.ContinueOnFailure"/>
+    /// </summary>
+    internal class FailureTransferHolder
+    {
+        public List<TransferFailedEventArgs> FailedEvents { get; internal set; }
+
+        private FailureTransferHolder()
+        {
+            FailedEvents = new List<TransferFailedEventArgs>();
+        }
+
+        public FailureTransferHolder(ContainerTransferOptions options)
+            : this()
+        {
+            options.TransferFailed += AppendFailedArg;
+        }
+
+        public FailureTransferHolder(SingleTransferOptions options)
+            : this()
+        {
+            options.TransferFailed += AppendFailedArg;
+        }
+
+        public FailureTransferHolder(List<SingleTransferOptions> optionsList)
+            : this()
+        {
+            foreach (SingleTransferOptions options in optionsList)
+            {
+                options.TransferFailed += AppendFailedArg;
+            }
+        }
+        public FailureTransferHolder(List<ContainerTransferOptions> optionsList)
+            : this()
+        {
+            foreach (ContainerTransferOptions options in optionsList)
+            {
+                options.TransferFailed += AppendFailedArg;
+            }
+        }
+
+        private Task AppendFailedArg(TransferFailedEventArgs args)
+        {
+            FailedEvents.Add(args);
+            return Task.CompletedTask;
+        }
+
+        public void AssertFailureCheck()
+        {
+            Assert.Multiple(() =>
+            {
+                foreach (TransferFailedEventArgs failure in FailedEvents)
+                {
+                    Assert.Fail(
+                        $"Failure occurred at Transfer id: {failure.TransferId}.\n" +
+                        $"Source Resource Path: {failure.SourceResource.Path}\n" +
+                        $"Destination Resource Path: {failure.DestinationResource.Path}\n" +
+                        $"Exception Message: {failure.Exception.Message}\n" +
+                        $"Exception Stack: {failure.Exception.StackTrace}\n");
+                }
+            });
+        }
+    }
+}