[Event Hubs Client] Capture Unobserved AMQP Link Close Errors (Azure#21426)

The AMQP transport library used by Event Hubs manages aspects of the transport
in the background, notably the flow of data into the prefetch queue.  In doing
so it controls attempts to read data, and detecting connection/link failures to
provide resiliency for callers.

Access to the AMQP link is managed through the `FaultTolerantAmqpObject<T>` which
uses a `GetOrCreate` pattern to handle faults and provide a resilient experience
for callers.  When a link was discarded due to a fault, it was not observable to
clients and a new link would be created on the next `GetOrCreate` call.

In the majority of scenarios, this behavior is desirable.  However, in some
scenarios - such as a partition being stolen by another event consumer asserting
exclusive access, the failure must be surfaced to callers so that the correct
remidiation can be taken.

The focus of these changes is to ensure that errors causing link termination
are surfaced via logging and that a stolen partition is treated as a special
case and is always surfaced to callers.

Because this behavior was a direct impact to the load balancing aspects of
the event processor, testing exposed some additional load balancing gaps that
were also addressed.  Lost partitions will now be detected more reliably and
the processor will proactively relinquish ownership and allow load balancing
to make recovery decisions.

Author: jsquire

sdk/eventhub/Azure.Messaging.EventHubs.Processor/CHANGELOG.md

@@ -2,8 +2,6 @@
 
 ## 5.5.0-beta.1 (Unreleased)
 
-## 5.5.0-beta.1 (Unreleased)
-
 ### Acknowledgments
 
 Thank you to our developer community members who helped to make the Event Hubs client libraries better with their contributions to this release:
@@ -14,7 +12,15 @@ Thank you to our developer community members who helped to make the Event Hubs c
 
 #### New Features
 
--  When stopping, the `EventProcessorClient` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout.  This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_
+-  When stopping, the `EventProcessorClient` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout.  This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_  
+
+- When the `EventProcessorClient` detects a partition being stolen outside of a load balancing cycle, it will immediately surrender ownership rather than waiting for a load balancing cycle to confirm the ownership change.  This will help reduce event duplication from overlapping ownership of processors.
+
+#### Key Bug Fixes
+
+- The `EventProcessorClient` will now properly respect another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background.  Previously, the client did not observe the error directly and attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed.
+
+- The  `EventProcessorClient` will now be less aggressive when considering whether or not to steal a partition, doing so only when it will correct an imbalance and preferring the status quo when the overall distribution would not change.  This will help reduce event duplication due to partitions moving between owners.
 
 ## 5.4.1 (2021-05-11)
 

sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Processor/PartitionLoadBalancer.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
@@ -99,7 +100,7 @@ internal class PartitionLoadBalancer
         ///   The set of partition ownership the associated event processor owns.  Partition ids are used as keys.
         /// </summary>
         ///
-        private Dictionary<string, EventProcessorPartitionOwnership> InstanceOwnership { get; set; } = new Dictionary<string, EventProcessorPartitionOwnership>();
+        private ConcurrentDictionary<string, EventProcessorPartitionOwnership> InstanceOwnership { get; set; } = new ConcurrentDictionary<string, EventProcessorPartitionOwnership>();
 
         /// <summary>
         ///   Initializes a new instance of the <see cref="PartitionLoadBalancer" /> class.
@@ -297,6 +298,16 @@ public virtual async Task RelinquishOwnershipAsync(CancellationToken cancellatio
             InstanceOwnership.Clear();
         }
 
+        /// <summary>
+        ///   Allows reporting that a partition was stolen by another event consumer causing ownership
+        ///   to be considered relinquished until the next load balancing cycle reconciles with persisted
+        ///   state.
+        /// </summary>
+        ///
+        /// <param name="partitionId">The identifier of the partition that was stolen.</param>
+        ///
+        public virtual void ReportPartitionStolen(string partitionId) => InstanceOwnership.TryRemove(partitionId, out _);
+
         /// <summary>
         ///   Finds and tries to claim an ownership if this processor instance is eligible to increase its ownership list.
         /// </summary>
@@ -389,9 +400,11 @@ public virtual async Task RelinquishOwnershipAsync(CancellationToken cancellatio
                     }
                 }
 
-                // If this processor has less than the minimum or any other processor has more than the maximum, then we need to steal a partition.
+                // If this processor has less than the minimum or it has less than the maximum at the same time another processor has more than the
+                // maximum, then we need to steal a partition.
 
-                if ((ownedPartitionsCount < minimumOwnedPartitionsCount) || (partitionsOwnedByProcessorWithGreaterThanMaximumOwnedPartitionsCount.Count > 0))
+                if ((ownedPartitionsCount < minimumOwnedPartitionsCount)
+                    || (ownedPartitionsCount < maximumOwnedPartitionsCount && partitionsOwnedByProcessorWithGreaterThanMaximumOwnedPartitionsCount.Count > 0))
                 {
                     Logger.ShouldStealPartition(OwnerIdentifier);
 
@@ -476,7 +489,7 @@ private async Task RenewOwnershipAsync(CancellationToken cancellationToken)
 
                 foreach (var oldOwnership in ownershipToRenew)
                 {
-                    InstanceOwnership.Remove(oldOwnership.PartitionId);
+                    InstanceOwnership.TryRemove(oldOwnership.PartitionId, out _);
                 }
 
                 foreach (var newOwnership in newOwnerships)
@@ -520,7 +533,6 @@ private async Task RenewOwnershipAsync(CancellationToken cancellationToken)
                                                                                                                             CancellationToken cancellationToken)
         {
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
-
             Logger.ClaimOwnershipStart(partitionId);
 
             // We need the eTag from the most recent ownership of this partition, even if it's expired.  We want to keep the offset and

sdk/eventhub/Azure.Messaging.EventHubs.Shared/tests/Processor/PartitionLoadBalancerTests.cs

@@ -914,6 +914,112 @@ public async Task RunLoadBalancingAsyncDoesNotRenewFreshPartitions()
             Assert.That(storageManager.TotalRenewals, Is.EqualTo(8), "There should be 4 initial claims and 4 renew claims");
         }
 
+        /// <summary>
+        ///   Verifies functionality of the <see cref="PartitionLoadBalancer" /> when a partition is
+        ///   reported stolen.
+        /// </summary>
+        ///
+        [Test]
+        public async Task ReportPartitionStolenAbandonsOwnership()
+        {
+            const int NumberOfPartitions = 3;
+
+            var partitionIds = Enumerable.Range(1, NumberOfPartitions).Select(p => p.ToString()).ToArray();
+            var storageManager = new InMemoryStorageManager();
+            var loadBalancer = new PartitionLoadBalancer(storageManager, Guid.NewGuid().ToString(), ConsumerGroup, FullyQualifiedNamespace, EventHubName, TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(10));
+
+            // Assume ownership of all partitions
+
+            await storageManager.ClaimOwnershipAsync(CreatePartitionOwnership(partitionIds, loadBalancer.OwnerIdentifier));
+            await loadBalancer.RunLoadBalancingAsync(partitionIds, CancellationToken.None);
+
+            Assert.That(loadBalancer.OwnedPartitionIds, Is.EquivalentTo(partitionIds), "The load balancer should own all partitions.");
+
+            // Report the first partition stolen and validate that it is immediately abandoned.
+
+            var firstPartition = partitionIds.First();
+            partitionIds = partitionIds.Skip(1).ToArray();
+            Assert.That(partitionIds, Does.Not.Contain(firstPartition), "The first partition should no longer exist in the set of ids.");
+
+            loadBalancer.ReportPartitionStolen(firstPartition);
+            Assert.That(loadBalancer.OwnedPartitionIds, Does.Not.Contain(firstPartition), "The load balancer should not own the first partition after it was stolen.");
+            Assert.That(loadBalancer.OwnedPartitionIds, Is.EquivalentTo(partitionIds), "The load balancer should own all but the first partition.");
+        }
+
+        /// <summary>
+        ///   Verifies functionality of the <see cref="PartitionLoadBalancer" /> when a partition is
+        ///   reported stolen.
+        /// </summary>
+        ///
+        [Test]
+        public async Task LoadBalancerDoesNotReclaimStolenPartitionIfStorageAgrees()
+        {
+            const int NumberOfPartitions = 3;
+
+            var partitionIds = Enumerable.Range(1, NumberOfPartitions).Select(p => p.ToString()).ToArray();
+            var storageManager = new InMemoryStorageManager();
+            var loadBalancer = new PartitionLoadBalancer(storageManager, Guid.NewGuid().ToString(), ConsumerGroup, FullyQualifiedNamespace, EventHubName, TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(10));
+
+            // Assume ownership of all partitions
+
+            await storageManager.ClaimOwnershipAsync(CreatePartitionOwnership(partitionIds, loadBalancer.OwnerIdentifier));
+            await loadBalancer.RunLoadBalancingAsync(partitionIds, CancellationToken.None);
+
+            Assert.That(loadBalancer.OwnedPartitionIds, Is.EquivalentTo(partitionIds), "The load balancer should own all partitions.");
+
+            // Report the first partition stolen and validate that it is immediately abandoned.
+
+            var firstPartition = partitionIds.First();
+
+            loadBalancer.ReportPartitionStolen(firstPartition);
+            Assert.That(loadBalancer.OwnedPartitionIds, Does.Not.Contain(firstPartition), "The load balancer should not own the first partition after it was stolen.");
+
+            // Update storage to reflect that the first partition is not owned.
+
+            var ownership = (await storageManager.ListOwnershipAsync(FullyQualifiedNamespace, EventHubName, ConsumerGroup)).Single(item => item.PartitionId == firstPartition);
+            ownership.OwnerIdentifier = "another-processor";
+
+            await storageManager.ClaimOwnershipAsync(new[] { ownership });
+            await loadBalancer.RunLoadBalancingAsync(partitionIds, CancellationToken.None);
+
+            Assert.That(loadBalancer.OwnedPartitionIds, Does.Not.Contain(firstPartition), "The load balancer should not own the first partition after load balancing.");
+        }
+
+        /// <summary>
+        ///   Verifies functionality of the <see cref="PartitionLoadBalancer" /> when a partition is
+        ///   reported stolen.
+        /// </summary>
+        ///
+        [Test]
+        public async Task LoadBalancerReclaimsStolenPartitionIfStorageDisagrees()
+        {
+            const int NumberOfPartitions = 3;
+
+            var partitionIds = Enumerable.Range(1, NumberOfPartitions).Select(p => p.ToString()).ToArray();
+            var storageManager = new InMemoryStorageManager();
+            var loadBalancer = new PartitionLoadBalancer(storageManager, Guid.NewGuid().ToString(), ConsumerGroup, FullyQualifiedNamespace, EventHubName, TimeSpan.FromMinutes(1), TimeSpan.FromSeconds(10));
+
+            // Assume ownership of all partitions
+
+            await storageManager.ClaimOwnershipAsync(CreatePartitionOwnership(partitionIds, loadBalancer.OwnerIdentifier));
+            await loadBalancer.RunLoadBalancingAsync(partitionIds, CancellationToken.None);
+
+            Assert.That(loadBalancer.OwnedPartitionIds, Is.EquivalentTo(partitionIds), "The load balancer should own all partitions.");
+
+            // Report the first partition stolen and validate that it is immediately abandoned.
+
+            var firstPartition = partitionIds.First();
+
+            loadBalancer.ReportPartitionStolen(firstPartition);
+            Assert.That(loadBalancer.OwnedPartitionIds, Does.Not.Contain(firstPartition), "The load balancer should not own the first partition after it was stolen.");
+
+            // Storage still reflects ownership of the stolen partition; running load balancing should
+            // reclaim it.
+
+            await loadBalancer.RunLoadBalancingAsync(partitionIds, CancellationToken.None);
+            Assert.That(loadBalancer.OwnedPartitionIds, Is.EquivalentTo(partitionIds), "The load balancer should own all partitions after considering storage.");
+        }
+
         /// <summary>
         ///   Creates a collection of <see cref="PartitionOwnership" /> based on the specified arguments.
         /// </summary>

sdk/eventhub/Azure.Messaging.EventHubs/CHANGELOG.md

@@ -14,6 +14,20 @@ Thank you to our developer community members who helped to make the Event Hubs c
 
 -  When stopping, the `EventProcessor<TPartition>` will now attempt to force-close the connection to the Event Hubs service to abort in-process read operations blocked on their timeout.  This should significantly help reduce the amount of time the processor takes to stop in many scenarios. _(Based on a community prototype contribution, courtesy of [danielmarbach](https://github.com/danielmarbach))_
 
+- When the `EventProcessor<TPartition>` detects a partition being stolen outside of a load balancing cycle, it will immediately surrender ownership rather than waiting for a load balancing cycle to confirm the ownership change.  This will help reduce event duplication from overlapping ownership of processors.
+
+- The `EventProcessor<TPartition>` now exposes the `ListPartitionIdsAsync` method, allowing custom processors to control the set of partitions known to the processor.  This can be used to reduce complexity when a custom processor is directly assigned a set of partitions to process rather than using load balancing to control ownership.
+
+- Additional verbose logging has been added to allow monitoring of lower-level AMQP operations such as creating links, terminal exceptions that fault a link without an active operation, and when the service force-closes links.
+
+#### Key Bug Fixes
+
+- The `EventProcessor<TPartition>` will now properly respect another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background.  Previously, the client did not observe the error directly and attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed.
+
+- The `EventProcessor<TPartition>` will now be less aggressive when considering whether or not to steal a partition, doing so only when it will correct an imbalance and preferring the status quo when the overall distribution would not change.  This will help reduce event duplication due to partitions moving between owners.
+
+- The `EventHubConsumerClient` and `PartitionReceiver` will now properly surface an exception when another another consumer stealing ownership of a partition when the service forcibly terminates the active link in the background.  Previously, the client did not observe the error directly and did not make callers attempted to recover the faulted link which reasserted ownership and caused the partition to "bounce" between owners until a load balancing cycle completed.
+
 ## 5.4.1 (2021-05-11)
 
 ### Changes

sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs

@@ -367,6 +367,7 @@ protected EventProcessor(int eventBatchMaximumCount, string consumerGroup, strin
         public override int GetHashCode() { throw null; }
         protected abstract System.Threading.Tasks.Task<System.Collections.Generic.IEnumerable<Azure.Messaging.EventHubs.Primitives.EventProcessorCheckpoint>> ListCheckpointsAsync(System.Threading.CancellationToken cancellationToken);
         protected abstract System.Threading.Tasks.Task<System.Collections.Generic.IEnumerable<Azure.Messaging.EventHubs.Primitives.EventProcessorPartitionOwnership>> ListOwnershipAsync(System.Threading.CancellationToken cancellationToken);
+        protected virtual System.Threading.Tasks.Task<string[]> ListPartitionIdsAsync(Azure.Messaging.EventHubs.EventHubConnection connection, System.Threading.CancellationToken cancellationToken) { throw null; }
         protected virtual System.Threading.Tasks.Task OnInitializingPartitionAsync(TPartition partition, System.Threading.CancellationToken cancellationToken) { throw null; }
         protected virtual System.Threading.Tasks.Task OnPartitionProcessingStoppedAsync(TPartition partition, Azure.Messaging.EventHubs.Processor.ProcessingStoppedReason reason, System.Threading.CancellationToken cancellationToken) { throw null; }
         protected abstract System.Threading.Tasks.Task OnProcessingErrorAsync(System.Exception exception, TPartition partition, string operationDescription, System.Threading.CancellationToken cancellationToken);

sdk/eventhub/Azure.Messaging.EventHubs/src/Amqp/AmqpClient.cs

@@ -182,6 +182,7 @@ protected AmqpClient(string host,
                     {
                         link.Session?.SafeClose();
                         link.SafeClose();
+                        EventHubsEventSource.Log.FaultTolerantAmqpObjectClose(nameof(RequestResponseAmqpLink), "", EventHubName, "", "", link.TerminalException?.Message);
                     });
             }
             finally
@@ -438,6 +439,7 @@ public override TransportProducer CreateProducer(string partitionId,
         /// <param name="eventPosition">The position within the partition where the consumer should begin reading events.</param>
         /// <param name="retryPolicy">The policy which governs retry behavior and try timeouts.</param>
         /// <param name="trackLastEnqueuedEventProperties">Indicates whether information on the last enqueued event on the partition is sent as events are received.</param>
+        /// <param name="invalidateConsumerWhenPartitionStolen">Indicates whether or not the consumer should consider itself invalid when a partition is stolen.</param>
         /// <param name="ownerLevel">The relative priority to associate with the link; for a non-exclusive link, this value should be <c>null</c>.</param>
         /// <param name="prefetchCount">Controls the number of events received and queued locally without regard to whether an operation was requested.  If <c>null</c> a default will be used.</param>
         /// <param name="prefetchSizeInBytes">The cache size of the prefetch queue. When set, the link makes a best effort to ensure prefetched messages fit into the specified size.</param>
@@ -449,6 +451,7 @@ public override TransportConsumer CreateConsumer(string consumerGroup,
                                                          EventPosition eventPosition,
                                                          EventHubsRetryPolicy retryPolicy,
                                                          bool trackLastEnqueuedEventProperties,
+                                                         bool invalidateConsumerWhenPartitionStolen,
                                                          long? ownerLevel,
                                                          uint? prefetchCount,
                                                          long? prefetchSizeInBytes)
@@ -462,6 +465,7 @@ public override TransportConsumer CreateConsumer(string consumerGroup,
                 partitionId,
                 eventPosition,
                 trackLastEnqueuedEventProperties,
+                invalidateConsumerWhenPartitionStolen,
                 ownerLevel,
                 prefetchCount,
                 prefetchSizeInBytes,