[Event Hubs] Enrich processor log (Azure#35169)

* [Event Hubs] Enrich processor log

The focus of these changes is to add the source of the event position chosen
for a partition reader to the log, to help troubleshooting to identify whether
a checkpoint was found or a default was used.

* Resolving reference issue and applying feedback

Author: jsquire

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

@@ -12,6 +12,8 @@
 
 ### Other Changes
 
+- Enhanced the log emitted when an event processor begins reading from a partition to report whether the offset chosen was based on a checkpoint or default value.
+
 ## 5.8.1 (2023-03-09)
 
 ### Bugs Fixed

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

@@ -14,6 +14,8 @@
 
 ### Other Changes
 
+- Enhanced the log emitted when an event processor begins reading from a partition to report whether the offset chosen was based on a checkpoint or default value.
+
 ## 5.8.1 (2023-03-09)
 
 ### Other Changes

sdk/eventhub/Azure.Messaging.EventHubs/src/Diagnostics/EventHubsEventSource.cs

@@ -2144,17 +2144,20 @@ public virtual void EventProcessorHighPartitionOwnershipWarning(string identifie
         /// <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="startingPosition">The position in the event stream that reading will start from.</param>
+        /// <param name="checkpointUsed"><c>true</c> if a checkpoint was used for the position; otherwise, <c>false</c>.</param>
         ///
-        [Event(105, Level = EventLevel.Verbose, Message = "The processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3} is initializing partition '{0}' with starting position: [{4}]")]
+        [Event(105, Level = EventLevel.Verbose, Message = "The processor instance with identifier '{1}' for Event Hub: {2} and Consumer Group: {3} is initializing partition '{0}' with starting position: [{4}]. Position chosen by {5}.")]
         public virtual void EventProcessorPartitionProcessingEventPositionDetermined(string partitionId,
                                                                                      string identifier,
                                                                                      string eventHubName,
                                                                                      string consumerGroup,
-                                                                                     string startingPosition)
+                                                                                     string startingPosition,
+                                                                                     bool checkpointUsed)
         {
             if (IsEnabled())
             {
-                WriteEvent(105, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, startingPosition ?? string.Empty);
+                var selectionBasedOn = checkpointUsed ? "checkpoint" : "default";
+                WriteEvent(105, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, startingPosition ?? string.Empty, selectionBasedOn);
             }
         }
 

sdk/eventhub/Azure.Messaging.EventHubs/src/Primitives/EventProcessor.cs

@@ -743,13 +743,13 @@ async Task performProcessing()
                 // was passed.  In the event that initialization is run and encounters an
                 // exception, it takes responsibility for firing the error handler.
 
-                var startingPosition = startingPositionOverride switch
+                var (startingPosition, checkpointUsed) = startingPositionOverride switch
                 {
-                    _ when startingPositionOverride.HasValue => startingPositionOverride.Value,
+                    _ when startingPositionOverride.HasValue => (startingPositionOverride.Value, false),
                     _ => await InitializePartitionForProcessingAsync(partition, cancellationSource.Token).ConfigureAwait(false)
                 };
 
-                Logger.EventProcessorPartitionProcessingEventPositionDetermined(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, startingPosition.ToString());
+                Logger.EventProcessorPartitionProcessingEventPositionDetermined(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, startingPosition.ToString(), checkpointUsed);
 
                 // Create the connection to be used for spawning consumers; if the creation
                 // fails, then consider the processing task to be failed.  The main processing
@@ -1795,15 +1795,15 @@ await Task.WhenAll(LoadBalancer.OwnedPartitionIds
         /// <param name="partition">The partition to initialize.</param>
         /// <param name="cancellationToken">A <see cref="CancellationToken"/> instance to signal the request to cancel the operation.</param>
         ///
-        /// <returns>The <see cref="EventPosition" /> to start processing from.</returns>
+        /// <returns>A tuple containing the <see cref="EventPosition" /> to start processing from and whether the position was based on a checkpoint or not.</returns>
         ///
         /// <remarks>
         ///   This method will invoke the error handler should an exception be encountered; the
         ///   exception will then be bubbled to callers.
         /// </remarks>
         ///
-        private async Task<EventPosition> InitializePartitionForProcessingAsync(TPartition partition,
-                                                                                CancellationToken cancellationToken)
+        private async Task<(EventPosition Position, bool CheckpointUsed)> InitializePartitionForProcessingAsync(TPartition partition,
+                                                                                                                CancellationToken cancellationToken)
         {
             var operationDescription = Resources.OperationClaimOwnership;
 
@@ -1824,10 +1824,10 @@ private async Task<EventPosition> InitializePartitionForProcessingAsync(TPartiti
 
                 if (checkpoint != null)
                 {
-                    return checkpoint.StartingPosition;
+                    return (checkpoint.StartingPosition, true);
                 }
 
-                return Options.DefaultStartingPosition;
+                return (Options.DefaultStartingPosition, false);
             }
             catch (Exception ex)
             {

sdk/eventhub/Azure.Messaging.EventHubs/stress/src/Azure.Messaging.EventHubs.Stress.csproj

@@ -8,7 +8,6 @@
 
   <ItemGroup>
     <PackageReference Include="System.Memory.Data" />
-    <PackageReference Include="Azure.Storage.Blobs" />
     <PackageReference Include="Microsoft.ApplicationInsights" />
     <PackageReference Include="CommandLineParser" />
   </ItemGroup>

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

@@ -1628,15 +1628,15 @@ public async Task CreatePartitionProcessorProcessingTaskLogsTheStartingPosition(
             cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);
 
             var partition = new EventProcessorPartition { PartitionId = "4" };
-            var expectedEevntPosition = EventPosition.FromSequenceNumber(332);
+            var expectedEventPosition = EventPosition.FromSequenceNumber(332);
             var completionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
             var mockLogger = new Mock<EventHubsEventSource>();
             var mockConnection = Mock.Of<EventHubConnection>();
             var mockConsumer = new Mock<SettableTransportConsumer>();
             var mockProcessor = new Mock<EventProcessor<EventProcessorPartition>>(5, "consumerGroup", "namespace", "eventHub", Mock.Of<TokenCredential>(), new EventProcessorOptions()) { CallBase = true };
 
             mockLogger
-                .Setup(log => log.EventProcessorPartitionProcessingEventPositionDetermined(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>()))
+                .Setup(log => log.EventProcessorPartitionProcessingEventPositionDetermined(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<bool>()))
                 .Callback(() => completionSource.TrySetResult(true));
 
             mockConsumer
@@ -1653,7 +1653,7 @@ public async Task CreatePartitionProcessorProcessingTaskLogsTheStartingPosition(
                 .Setup(processor => processor.CreateConsumer(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<EventPosition>(), mockConnection, It.IsAny<EventProcessorOptions>(), It.IsAny<bool>()))
                 .Returns(mockConsumer.Object);
 
-            var partitionProcessor = mockProcessor.Object.CreatePartitionProcessor(partition, cancellationSource, expectedEevntPosition);
+            var partitionProcessor = mockProcessor.Object.CreatePartitionProcessor(partition, cancellationSource, expectedEventPosition);
 
             await Task.WhenAny(completionSource.Task, Task.Delay(Timeout.Infinite, cancellationSource.Token));
             Assert.That(cancellationSource.IsCancellationRequested, Is.False, "The cancellation token should not have been signaled.");
@@ -1664,7 +1664,8 @@ public async Task CreatePartitionProcessorProcessingTaskLogsTheStartingPosition(
                     mockProcessor.Object.Identifier,
                     mockProcessor.Object.EventHubName,
                     mockProcessor.Object.ConsumerGroup,
-                    expectedEevntPosition.ToString()),
+                    expectedEventPosition.ToString(),
+                    false),
                 Times.Once);
 
             cancellationSource.Cancel();