[Event Hubs] Log improvements + Amqp version bump (Azure#38319)

* [Event Hubs] Log improvements + Amqp version bump

The focus of these changes is to improve logs for the event processor
types and AMQP consumer.  The most notable change is the addition of the
first and last sequence number of the batch when reading from Event Hubs
and dispatching events to be processed.

The version of the AMQP library has been bumped to the latest stable (2.6.3),
which includes a fix for timer duration calculations during link creation
and a small handful of efficiency improvements.

Author: jsquire

eng/Packages.Data.props

@@ -105,7 +105,7 @@
     <PackageReference Update="Azure.ResourceManager" Version="1.7.0" />
 
     <!-- Other approved packages -->
-    <PackageReference Update="Microsoft.Azure.Amqp" Version="2.6.2" />
+    <PackageReference Update="Microsoft.Azure.Amqp" Version="2.6.3" />
     <PackageReference Update="Microsoft.Azure.WebPubSub.Common" Version="1.2.0" />
     <PackageReference Update="Microsoft.Identity.Client" Version="4.54.1" />
     <PackageReference Update="Microsoft.Identity.Client.Extensions.Msal" Version="2.31.0" />

sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/Diagnostics/EventProcessorClientEventSource.cs

@@ -48,16 +48,18 @@ protected EventProcessorClientEventSource() : base(EventSourceName)
         /// <param name="identifier">A unique name used to identify the event processor.</param>
         /// <param name="eventHubName">The name of the Event Hub that the processor is associated with.</param>
         /// <param name="consumerGroup">The name of the consumer group that the processor is associated with.</param>
+        /// <param name="operationId">An identifier for the processing operation, allowing its activities to be correlated.</param>
         ///
-        [Event(20, Level = EventLevel.Verbose, Message = "Starting to process a batch of events for partition '{0}' by processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}.")]
+        [Event(20, Level = EventLevel.Verbose, Message = "Starting to process a batch of events for partition '{0}' by processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}.  Operation Id: '{4}'")]
         public virtual void EventBatchProcessingStart(string partitionId,
                                                       string identifier,
                                                       string eventHubName,
-                                                      string consumerGroup)
+                                                      string consumerGroup,
+                                                      string operationId)
         {
             if (IsEnabled())
             {
-                WriteEvent(20, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty);
+                WriteEvent(20, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, operationId ?? string.Empty);
             }
         }
 
@@ -69,16 +71,18 @@ public virtual void EventBatchProcessingStart(string partitionId,
         /// <param name="identifier">A unique name used to identify the event processor.</param>
         /// <param name="eventHubName">The name of the Event Hub that the processor is associated with.</param>
         /// <param name="consumerGroup">The name of the consumer group that the processor is associated with.</param>
+        /// <param name="operationId">An identifier for the processing operation, allowing its activities to be correlated.</param>
         ///
-        [Event(21, Level = EventLevel.Verbose, Message = "Completed processing a batch of events for partition '{0}' by processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}.")]
+        [Event(21, Level = EventLevel.Verbose, Message = "Completed processing a batch of events for partition '{0}' by processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}.  Operation Id: '{4}'")]
         public virtual void EventBatchProcessingComplete(string partitionId,
                                                          string identifier,
                                                          string eventHubName,
-                                                         string consumerGroup)
+                                                         string consumerGroup,
+                                                         string operationId)
         {
             if (IsEnabled())
             {
-                WriteEvent(21, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty);
+                WriteEvent(21, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, operationId ?? string.Empty);
             }
         }
 
@@ -90,18 +94,20 @@ public virtual void EventBatchProcessingComplete(string partitionId,
         /// <param name="identifier">A unique name used to identify the event processor.</param>
         /// <param name="eventHubName">The name of the Event Hub that the processor is associated with.</param>
         /// <param name="consumerGroup">The name of the consumer group that the processor is associated with.</param>
+        /// <param name="operationId">An identifier for the processing operation, allowing its activities to be correlated.</param>
         /// <param name="errorMessage">The message for the exception that occurred.</param>
         ///
-        [Event(22, Level = EventLevel.Error, Message = "An exception occurred while processing events for partition '{0}' by processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}.  Error Message: '{4}'")]
+        [Event(22, Level = EventLevel.Error, Message = "An exception occurred while processing events for partition '{0}' by processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3}.  Operation Id: '{5}'; Error Message: '{4}'")]
         public virtual void EventBatchProcessingError(string partitionId,
                                                       string identifier,
                                                       string eventHubName,
                                                       string consumerGroup,
+                                                      string operationId,
                                                       string errorMessage)
         {
             if (IsEnabled())
             {
-                WriteEvent(22, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, errorMessage ?? string.Empty);
+                WriteEvent(22, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, errorMessage ?? string.Empty, operationId ?? string.Empty);
             }
         }
 
@@ -191,6 +197,31 @@ public virtual void ValidationCleanupError(string identifier,
             }
         }
 
+        /// <summary>
+        ///   Indicates that an <see cref="EventProcessorClient" /> has begin processing a batch of events for a partition.
+        /// </summary>
+        ///
+        /// <param name="sequenceNumber">The sequence number of the event being passed to the handler for processing.</param>
+        /// <param name="partitionId">The identifier of the Event Hub partition whose processing is taking place.</param>
+        /// <param name="identifier">A unique name used to identify the event processor.</param>
+        /// <param name="eventHubName">The name of the Event Hub that the processor is associated with.</param>
+        /// <param name="consumerGroup">The name of the consumer group that the processor is associated with.</param>
+        /// <param name="operationId">An identifier for the processing operation, allowing its activities to be correlated.</param>
+        ///
+        [Event(27, Level = EventLevel.Verbose, Message = "Invoking the event processing handler for sequence number '{0}'.  Partition '{1}'; Processor Identifier '{2}'; Event Hub: {3}; Consumer Group: {4}; Operation Id: '{5}'")]
+        public virtual void EventBatchProcessingHandlerCall(string sequenceNumber,
+                                                            string partitionId,
+                                                            string identifier,
+                                                            string eventHubName,
+                                                            string consumerGroup,
+                                                            string operationId)
+        {
+            if (IsEnabled())
+            {
+                WriteEvent(27, sequenceNumber ?? string.Empty, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, operationId ?? string.Empty);
+            }
+        }
+
         /// <summary>
         ///   Writes an event with four string arguments into a stack allocated <see cref="EventSource.EventData"/> struct
         ///   to avoid the parameter array allocation on the WriteEvent methods.
@@ -280,5 +311,59 @@ private unsafe void WriteEvent(int eventId,
                 WriteEventCore(eventId, 5, eventPayload);
             }
         }
+
+        /// <summary>
+        ///   Writes an event with five string arguments into a stack allocated
+        ///   <see cref="EventSource.EventData"/> struct to avoid the parameter array allocation on the WriteEvent methods.
+        /// </summary>
+        ///
+        /// <param name="eventId">The identifier of the event.</param>
+        /// <param name="arg1">The first argument.</param>
+        /// <param name="arg2">The second argument.</param>
+        /// <param name="arg3">The third argument.</param>
+        /// <param name="arg4">The fourth argument.</param>
+        /// <param name="arg5">The fifth argument.</param>
+        /// <param name="arg6">The sixth argument.</param>
+        ///
+        [NonEvent]
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private unsafe void WriteEvent(int eventId,
+                                       string arg1,
+                                       string arg2,
+                                       string arg3,
+                                       string arg4,
+                                       string arg5,
+                                       string arg6)
+        {
+            fixed (char* arg1Ptr = arg1)
+            fixed (char* arg2Ptr = arg2)
+            fixed (char* arg3Ptr = arg3)
+            fixed (char* arg4Ptr = arg4)
+            fixed (char* arg5Ptr = arg5)
+            fixed (char* arg6Ptr = arg6)
+            {
+                var eventPayload = stackalloc EventData[6];
+
+                eventPayload[0].Size = (arg1.Length + 1) * sizeof(char);
+                eventPayload[0].DataPointer = (IntPtr)arg1Ptr;
+
+                eventPayload[1].Size = (arg2.Length + 1) * sizeof(char);
+                eventPayload[1].DataPointer = (IntPtr)arg2Ptr;
+
+                eventPayload[2].Size = (arg3.Length + 1) * sizeof(char);
+                eventPayload[2].DataPointer = (IntPtr)arg3Ptr;
+
+                eventPayload[3].Size = (arg4.Length + 1) * sizeof(char);
+                eventPayload[3].DataPointer = (IntPtr)arg4Ptr;
+
+                eventPayload[4].Size = (arg5.Length + 1) * sizeof(char);
+                eventPayload[4].DataPointer = (IntPtr)arg5Ptr;
+
+                eventPayload[5].Size = (arg5.Length + 1) * sizeof(char);
+                eventPayload[5].DataPointer = (IntPtr)arg5Ptr;
+
+                WriteEventCore(eventId, 6, eventPayload);
+            }
+        }
     }
 }

sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/EventProcessorClient.cs

@@ -1015,14 +1015,15 @@ protected override async Task OnProcessingEventBatchAsync(IEnumerable<EventData>
         {
             cancellationToken.ThrowIfCancellationRequested<TaskCanceledException>();
 
+            var operation = Guid.NewGuid().ToString("D", CultureInfo.InvariantCulture);
             var context = default(PartitionContext);
             var eventArgs = default(ProcessEventArgs);
             var caughtExceptions = default(List<Exception>);
             var emptyBatch = true;
 
             try
             {
-                Logger.EventBatchProcessingStart(partition.PartitionId, Identifier, EventHubName, ConsumerGroup);
+                Logger.EventBatchProcessingStart(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, operation);
 
                 // Attempt to process each event in the batch, marking if the batch was non-empty.  Exceptions during
                 // processing should be logged and cached, as the batch must be processed completely to avoid losing events.
@@ -1043,6 +1044,8 @@ protected override async Task OnProcessingEventBatchAsync(IEnumerable<EventData>
 
                     try
                     {
+                        Logger.EventBatchProcessingHandlerCall(eventData.SequenceNumber.ToString(), partition.PartitionId, Identifier, EventHubName, ConsumerGroup, operation);
+
                         context ??= new ProcessorPartitionContext(FullyQualifiedNamespace, EventHubName, ConsumerGroup, partition.PartitionId, () => ReadLastEnqueuedEventProperties(partition.PartitionId));
                         eventArgs = new ProcessEventArgs(context, eventData, updateToken => UpdateCheckpointAsync(partition.PartitionId, eventData.Offset, eventData.SequenceNumber, updateToken), cancellationToken);
 
@@ -1053,7 +1056,7 @@ protected override async Task OnProcessingEventBatchAsync(IEnumerable<EventData>
                         // This exception is not surfaced to the error handler or bubbled, as the entire batch must be
                         // processed or events will be lost.  Preserve the exceptions, should any occur.
 
-                        Logger.EventBatchProcessingError(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, ex.Message);
+                        Logger.EventBatchProcessingError(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, operation, ex.Message);
 
                         caughtExceptions ??= new List<Exception>();
                         caughtExceptions.Add(ex);
@@ -1065,6 +1068,8 @@ protected override async Task OnProcessingEventBatchAsync(IEnumerable<EventData>
 
                 if (emptyBatch)
                 {
+                    Logger.EventBatchProcessingHandlerCall("<< No Event >>", partition.PartitionId, Identifier, EventHubName, ConsumerGroup, operation);
+
                     eventArgs = new ProcessEventArgs(new EmptyPartitionContext(FullyQualifiedNamespace, EventHubName, ConsumerGroup, partition.PartitionId), null, EmptyEventUpdateCheckpoint, cancellationToken);
                     await _processEventAsync(eventArgs).ConfigureAwait(false);
                 }
@@ -1074,12 +1079,12 @@ protected override async Task OnProcessingEventBatchAsync(IEnumerable<EventData>
                 // This exception was either not related to processing events or was the result of sending an empty batch to be
                 // processed.  Since there would be no other caught exceptions, tread this like a single case.
 
-                Logger.EventBatchProcessingError(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, ex.Message);
+                Logger.EventBatchProcessingError(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, operation, ex.Message);
                 throw;
             }
             finally
             {
-                Logger.EventBatchProcessingComplete(partition.PartitionId, Identifier, EventHubName, ConsumerGroup);
+                Logger.EventBatchProcessingComplete(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, operation);
             }
 
             // Deal with any exceptions that occurred while processing the batch.  If more than one was

sdk/eventhub/Azure.Messaging.EventHubs.Processor/tests/Processor/EventProcessorClientTests.cs

@@ -1130,15 +1130,17 @@ public async Task EventProcessingLogsExecution()
                     partitionId,
                     processorClient.Identifier,
                     processorClient.EventHubName,
-                    processorClient.ConsumerGroup),
+                    processorClient.ConsumerGroup,
+                    It.IsAny<string>()),
                 Times.Once);
 
             mockLogger
                 .Verify(log => log.EventBatchProcessingComplete(
                     partitionId,
                     processorClient.Identifier,
                     processorClient.EventHubName,
-                    processorClient.ConsumerGroup),
+                    processorClient.ConsumerGroup,
+                    It.IsAny<string>()),
                 Times.Once);
 
             cancellationSource.Cancel();
@@ -1176,6 +1178,7 @@ public async Task EventProcessingLogsExceptions()
                     processorClient.Identifier,
                     processorClient.EventHubName,
                     processorClient.ConsumerGroup,
+                    It.IsAny<string>(),
                     expectedException.Message),
                 Times.Exactly(eventBatch.Length));
 

sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/Diagnostics/PartitionLoadBalancerEventSource.cs

@@ -136,14 +136,18 @@ public virtual void ShouldStealPartition(string identifier)
         ///   Indicates that stealable partitions were found, so randomly picking one of them to claim.
         /// </summary>
         ///
+        /// <param name="partitionId">The identifier of the partition that was selected to be stolen.</param>
+        /// <param name="stolenFrom">The identifier of the event processor that is being stolen from.</param>
         /// <param name="identifier">A unique name used to identify the associated event processor.</param>
         ///
-        [Event(7, Level = EventLevel.Informational, Message = "No unclaimed partitions, stealing from another event processor. (Identifier: '{0}')")]
-        public virtual void StealPartition(string identifier)
+        [Event(7, Level = EventLevel.Informational, Message = "No unclaimed partitions, attempting to steal partition '{0}' from event processor '{1}'. (Identifier: '{2}')")]
+        public virtual void StealPartition(string partitionId,
+                                           string stolenFrom,
+                                           string identifier)
         {
             if (IsEnabled())
             {
-                WriteEvent(7, identifier ?? string.Empty);
+                WriteEvent(7, partitionId ?? string.Empty, stolenFrom ?? string.Empty, identifier ?? string.Empty);
             }
         }