[Event Hubs] Detect offset placeholder value (Azure#49319)

* [Event Hubs] Detect offset placeholder value

The focus of these changes is to ensure that the
placeholder value used to represent missing offsets
in v5.11.3 - v5.11.6 is properly detected when
reading a checkpoint and not used as a valid offset
when reading from a partition.

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

Author: jsquire

sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/BlobCheckpointStore/BlobCheckpointStoreInternal.cs

@@ -25,6 +25,9 @@ namespace Azure.Messaging.EventHubs.Primitives
     ///
     internal partial class BlobCheckpointStoreInternal : CheckpointStore
     {
+        /// <summary>The text used as a placeholder when no offset metadata exists in some older checkpoint formats.</summary>
+        internal const string NoOffsetPlaceholderText = "no offset";
+
 #pragma warning disable CA1802 // Use a constant field
         /// <summary>A message to use when throwing exception when checkpoint container or blob does not exists.</summary>
         private static readonly string BlobsResourceDoesNotExist = "The Azure Storage Blobs container or blob used by the Event Processor Client does not exist.";
@@ -524,7 +527,7 @@ private EventProcessorCheckpoint CreateCheckpoint(string fullyQualifiedNamespace
             var sequenceNumber = default(long?);
             var clientIdentifier = default(string);
 
-            if (metadata.TryGetValue(BlobMetadataKey.Offset, out var offsetStr) && !string.IsNullOrEmpty(offsetStr))
+            if (metadata.TryGetValue(BlobMetadataKey.Offset, out var offsetStr) && !IsPlaceholderOffset(offsetStr))
             {
                 offset = offsetStr;
                 startingPosition = EventPosition.FromOffset(offsetStr, false);
@@ -631,6 +634,20 @@ private async Task<EventProcessorCheckpoint> CreateLegacyCheckpoint(string fully
             };
         }
 
+        /// <summary>
+        ///   Determines whether an offset value is a placeholder slug that was written
+        ///   for legacy compatibility purposes.
+        /// </summary>
+        ///
+        /// <param name="offset">The offset to test.</param>
+        ///
+        /// <returns><c>true</c> if <paramref name="offset"/> is the placeholder offset; otherwise, <c>false</c>.</returns>
+        ///
+        /// <seealso href="https://github.com/Azure/azure-sdk-for-net/blob/Azure.Messaging.EventHubs_5.11.6/sdk/eventhub/Azure.Messaging.EventHubs.Shared/src/BlobCheckpointStore/BlobCheckpointStoreInternal.cs#L446" />
+        ///
+        private bool IsPlaceholderOffset(string offset) =>
+            string.IsNullOrEmpty(offset) || string.Equals(offset, NoOffsetPlaceholderText, StringComparison.OrdinalIgnoreCase);
+
         /// <summary>
         ///   Attempts to read a legacy checkpoint JSON format and extract an offset and a sequence number.
         /// </summary>

sdk/eventhub/Azure.Messaging.EventHubs.Shared/tests/BlobCheckpointStore/BlobsCheckpointStoreInternalTests.cs

@@ -358,6 +358,47 @@ public async Task GetCheckpointLogsStartAndComplete()
             mockLog.Verify(m => m.GetCheckpointComplete(FullyQualifiedNamespace, EventHubName, ConsumerGroup, partition, null, It.IsAny<DateTimeOffset>()));
         }
 
+        /// <summary>
+        ///   Verifies basic functionality of GetCheckpointAsync and ensures the starting position is set correctly.
+        /// </summary>
+        ///
+        [Test]
+        [TestCase(null)]
+        [TestCase("")]
+        [TestCase(BlobCheckpointStoreInternal.NoOffsetPlaceholderText)]
+        public async Task GetCheckpointIgnoresPlaceholderOffsets(string placeholderOffset)
+        {
+            var expectedSequence = 133;
+            var expectedStartingPosition = EventPosition.FromSequenceNumber(expectedSequence, false);
+            var partition = Guid.NewGuid().ToString();
+
+            var blobList = new List<BlobItem>
+            {
+                BlobsModelFactory.BlobItem($"{FullyQualifiedNamespaceLowercase}/{EventHubNameLowercase}/{ConsumerGroupLowercase}/checkpoint/{partition}",
+                                           false,
+                                           BlobsModelFactory.BlobItemProperties(true, lastModified: DateTime.UtcNow, eTag: new ETag(MatchingEtag)),
+                                           "snapshot",
+                                           new Dictionary<string, string>
+                                           {
+                                               {BlobMetadataKey.OwnerIdentifier, Guid.NewGuid().ToString()},
+                                               {BlobMetadataKey.Offset, placeholderOffset},
+                                               {BlobMetadataKey.SequenceNumber, expectedSequence.ToString()}
+                                           })
+            };
+
+            var target = new BlobCheckpointStoreInternal(new MockBlobContainerClient() { Blobs = blobList });
+            var checkpoint = await target.GetCheckpointAsync(FullyQualifiedNamespace, EventHubName, ConsumerGroup, partition, CancellationToken.None);
+
+            Assert.That(checkpoint, Is.Not.Null, "A checkpoint should have been returned.");
+            Assert.That(checkpoint.StartingPosition, Is.EqualTo(expectedStartingPosition));
+
+            Assert.That(checkpoint, Is.InstanceOf<BlobCheckpointStoreInternal.BlobStorageCheckpoint>(), "Checkpoint instance was not the expected type.");
+
+            var blobCheckpoint = (BlobCheckpointStoreInternal.BlobStorageCheckpoint)checkpoint;
+            Assert.That(blobCheckpoint.Offset, Is.Null, $"The offset should not have been populated, as it was not set.");
+            Assert.That(expectedSequence, Is.EqualTo(blobCheckpoint.SequenceNumber), "Checkpoint sequence number did not have the correct value.");
+        }
+
         /// <summary>
         ///   Verifies basic functionality of GetCheckpointAsync and ensures the starting position is set correctly.
         /// </summary>

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

@@ -1,14 +1,10 @@
 # Release History
 
-## 5.13.0-beta.1 (Unreleased)
-
-### Features Added
-
-### Breaking Changes
+## 5.12.1 (2025-04-09)
 
 ### Bugs Fixed
 
-### Other Changes
+- Fixed a bug which caused the placeholder value used to represent missing offsets in v5.11.3 - v5.11.6 to not be properly detected and incorrectly used as a valid offset when reading from a partition.
 
 ## 5.12.0 (2025-04-08)
 

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

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
     <Description>Azure Event Hubs is a highly scalable publish-subscribe service that can ingest millions of events per second and stream them to multiple consumers.  This client library allows for both publishing and consuming events using Azure Event Hubs. For more information about Event Hubs, see https://azure.microsoft.com/en-us/services/event-hubs/</Description>
-    <Version>5.13.0-beta.1</Version>
+    <Version>5.12.1</Version>
     <!--The ApiCompatVersion is managed automatically and should not generally be modified manually. -->
     <ApiCompatVersion>5.12.0</ApiCompatVersion>
     <PackageTags>Azure;Event Hubs;EventHubs;.NET;AMQP;IoT;$(PackageCommonTags)</PackageTags>