[Event Hubs] Flaky Test Fix (Azure#29740)

The focus of these changes is to fix a test believed to be flaky due to
timing issues around callback execution and ignoring of a cancellation token.

Author: jsquire

sdk/eventhub/Azure.Messaging.EventHubs/tests/Primitives/EventProcessorTests.StartStop.cs

@@ -637,15 +637,18 @@ public async Task StopProcessingStopsProcessing(bool async)
 
             mockConsumer
                 .Setup(consumer => consumer.ReceiveAsync(It.IsAny<int>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()))
-                .ReturnsAsync(new List<EventData> { new EventData(new BinaryData(Array.Empty<byte>())), new EventData(new BinaryData(Array.Empty<byte>())) });
+                .Returns<int, object, CancellationToken>((_, _, cancelToken) =>
+                {
+                    cancelToken.ThrowIfCancellationRequested<TaskCanceledException>();
+                    return Task.FromResult((IReadOnlyList<EventData>)new List<EventData> { new EventData(new BinaryData(Array.Empty<byte>())), new EventData(new BinaryData(Array.Empty<byte>())) });
+                });
 
             mockProcessor
                 .Setup(processor => processor.CreateConnection())
                 .Returns(mockConnection.Object);
 
             mockProcessor
                 .Setup(processor => processor.ProcessEventBatchAsync(It.IsAny<EventProcessorPartition>(), It.IsAny<IReadOnlyList<EventData>>(), It.IsAny<bool>(), It.IsAny<CancellationToken>()))
-                .Returns(Task.CompletedTask)
                 .Callback(() =>
                 {
                     if ((stopInvoked) && (!cancellationSource.Token.IsCancellationRequested))
@@ -654,7 +657,8 @@ public async Task StopProcessingStopsProcessing(bool async)
                     }
 
                     completionSource.TrySetResult(true);
-                });
+                })
+                .Returns(Task.CompletedTask);
 
             mockConnection
                 .Setup(connection => connection.CreateTransportConsumer(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<EventPosition>(), It.IsAny<EventHubsRetryPolicy>(), It.IsAny<bool>(), It.IsAny<bool>(), It.IsAny<long>(), It.IsAny<uint?>(), It.IsAny<long?>()))