[Event Hubs] Minor Fixes (#24590)

The focus of these changes is a set of minor fixes, including:

- Fixing the flaky `LoadBalancingAppliesTheGreedyStrategy` test
- Logging the selected event position when a partition is initialized
- Removing an outdated and incorrect set of remarks.

Author: jsquire

sdk/eventhub/Azure.Messaging.EventHubs/src/Consumer/EventPosition.cs

@@ -43,12 +43,6 @@ public struct EventPosition : IEquatable<EventPosition>
         ///
         /// <value>Expected to be <c>null</c> if the event position represents a sequence number or enqueue time.</value>
         ///
-        /// <remarks>
-        ///   The offset is the relative position for event in the context of the stream.  The offset
-        ///   should not be considered a stable value, as the same offset may refer to a different event
-        ///   as events reach the age limit for retention and are no longer visible within the stream.
-        /// </remarks>
-        ///
         internal string Offset { get; set; }
 
         /// <summary>

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

@@ -1619,5 +1619,28 @@ public virtual void EventProcessorHighPartitionOwnershipWarning(string identifie
                 WriteEvent(104, identifier ?? string.Empty, eventHubName ?? string.Empty, totalPartitionCount, ownedPartitionCount, maximumAdvisedCount);
             }
         }
+
+        /// <summary>
+        ///   Indicates that an <see cref="EventProcessor{TPartition}" /> instance has taken ownership of a partition and is actively processing it.
+        /// </summary>
+        ///
+        /// <param name="partitionId">The identifier of the Event Hub partition whose processing is starting.</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="startingPosition">The position in the event stream that reading will start from.</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}]")]
+        public virtual void EventProcessorPartitionProcessingEventPositionDetermined(string partitionId,
+                                                                                     string identifier,
+                                                                                     string eventHubName,
+                                                                                     string consumerGroup,
+                                                                                     string startingPosition)
+        {
+            if (IsEnabled())
+            {
+                WriteEvent(105, partitionId ?? string.Empty, identifier ?? string.Empty, eventHubName ?? string.Empty, consumerGroup ?? string.Empty, startingPosition ?? string.Empty);
+            }
+        }
     }
 }

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

@@ -690,6 +690,8 @@ async Task performProcessing()
                     _ => await InitializePartitionForProcessingAsync(partition, cancellationSource.Token).ConfigureAwait(false)
                 };
 
+                Logger.EventProcessorPartitionProcessingEventPositionDetermined(partition.PartitionId, Identifier, EventHubName, ConsumerGroup, startingPosition.ToString());
+
                 // Create the connection to be used for spawning consumers; if the creation
                 // fails, then consider the processing task to be failed.  The main processing
                 // loop will take responsibility for attempting to restart or relinquishing ownership.

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

@@ -1871,10 +1871,10 @@ public async Task LoadBalancingAppliesTheGreedyStrategy()
             Assert.That(ownedPartitions.Contains(firstPartiton), Is.True, "The first partition should be owned.");
             Assert.That(ownedPartitions.Contains(secondPartition), Is.True, "The second partition should be owned.");
 
-            // Due to non-determinism, the exact number of cycle is not known; allow for a small range.
+            // Due to non-determinism, the exact number of cycle is not known.
 
             mockLoadBalancer
-                .Verify(lb => lb.RunLoadBalancingAsync(partitionIds, It.IsAny<CancellationToken>()), Times.Between(3, 6, Moq.Range.Inclusive), "The load balancer did not run the correct number of cycles.");
+                .Verify(lb => lb.RunLoadBalancingAsync(partitionIds, It.IsAny<CancellationToken>()), Times.AtLeast(3), "The load balancer did not run the correct number of cycles.");
         }
 
         /// <summary>

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

@@ -1526,6 +1526,60 @@ public async Task CreatePartitionProcessorProcessingTaskStartsTheConsumerAtTheCo
                 Times.Once);
         }
 
+        /// <summary>
+        ///   Verifies functionality of the <see cref="EventProcessor{TPartition}.CreatePartitionProcessor" />
+        ///   method.
+        /// </summary>
+        ///
+        [Test]
+        public async Task CreatePartitionProcessorProcessingTaskLogsTheStartingPosition()
+        {
+            using var cancellationSource = new CancellationTokenSource();
+            cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);
+
+            var partition = new EventProcessorPartition { PartitionId = "4" };
+            var expectedEevntPosition = 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>()))
+                .Callback(() => completionSource.TrySetResult(true));
+
+            mockConsumer
+                .Setup(consumer => consumer.ReceiveAsync(It.IsAny<int>(), It.IsAny<TimeSpan?>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(EmptyBatch);
+
+            mockProcessor.Object.Logger = mockLogger.Object;
+
+            mockProcessor
+                .Setup(processor => processor.CreateConnection())
+                .Returns(mockConnection);
+
+            mockProcessor
+                .Setup(processor => processor.CreateConsumer(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<string>(), It.IsAny<EventPosition>(), mockConnection, It.IsAny<EventProcessorOptions>()))
+                .Returns(mockConsumer.Object);
+
+            var partitionProcessor = mockProcessor.Object.CreatePartitionProcessor(partition, cancellationSource, expectedEevntPosition);
+
+            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.");
+
+            mockLogger
+                .Verify(log => log.EventProcessorPartitionProcessingEventPositionDetermined(
+                    partition.PartitionId,
+                    mockProcessor.Object.Identifier,
+                    mockProcessor.Object.EventHubName,
+                    mockProcessor.Object.ConsumerGroup,
+                    expectedEevntPosition.ToString()),
+                Times.Once);
+
+            cancellationSource.Cancel();
+        }
+
         /// <summary>
         ///   Verifies functionality of the <see cref="EventProcessor{TPartition}.CreatePartitionProcessor" />
         ///   method.