Update Event Hubs offset handling to use string representation because of changes in Azure.Messaging.EventHubs 5.12.0+ (#55)

tokarchyn · web-flow · commit cb6f27c5a32c · 2025-07-10T17:15:41.000+02:00
diff --git a/.editorconfig b/.editorconfig
@@ -249,6 +249,3 @@ resharper_inconsistent_naming_highlighting = none
 
 # Solution-specific settings/overrides ------------------------------------------------------
 
-## TBD
-
-
diff --git a/Directory.Packages.props b/Directory.Packages.props
@@ -3,11 +3,11 @@
         <PackageVersion Include="Azure.Core" Version="1.39.0" />
         <PackageVersion Include="Azure.Data.Tables" Version="12.8.3" />
         <PackageVersion Include="Azure.Identity" Version="1.11.4" />
-        <PackageVersion Include="Azure.Messaging.EventHubs" Version="5.11.3" />
+        <PackageVersion Include="Azure.Messaging.EventHubs" Version="5.12.2" />
         <PackageVersion Include="Azure.Messaging.ServiceBus" Version="7.17.5" />
         <PackageVersion Include="Azure.ResourceManager.ServiceBus" Version="1.0.1" />
         <PackageVersion Include="Azure.ResourceManager.Storage" Version="1.3.0" />
-        <PackageVersion Include="Azure.ResourceManager.KeyVault" Version="1.3.0" />
+        <PackageVersion Include="Azure.ResourceManager.KeyVault" Version="1.3.0" />
         <PackageVersion Include="Azure.ResourceManager.EventHubs" Version="1.1.0" />
         <PackageVersion Include="Azure.Storage.Blobs" Version="12.20.0" />
         <PackageVersion Include="Azure.Security.KeyVault.Secrets" Version="4.6.0" />
diff --git a/Spotflow.InMemory.Azure.sln b/Spotflow.InMemory.Azure.sln
@@ -5,6 +5,7 @@ VisualStudioVersion = 17.8.34525.116
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{2C6BE026-65BE-4D7A-8E4D-F8CF8E2C300E}"
 	ProjectSection(SolutionItems) = preProject
+		.editorconfig = .editorconfig
 		Directory.Build.props = Directory.Build.props
 		Directory.Packages.props = Directory.Packages.props
 		LICENSE.md = LICENSE.md
diff --git a/src/Spotflow.InMemory.Azure.EventHubs/InMemoryPartitionReceiver.cs b/src/Spotflow.InMemory.Azure.EventHubs/InMemoryPartitionReceiver.cs
@@ -218,11 +218,13 @@ private async Task<IEnumerable<EventData>> ReceiveBatchCoreAsync(int maximumEven
                 }
                 else
                 {
+#pragma warning disable CS0618 // Type or member is obsolete
                     _lastEnqueuedEventProperties = new(
                         partitionProperties.LastEnqueuedSequenceNumber,
-                        partitionProperties.LastEnqueuedOffset,
+                        lastOffset: partitionProperties.LastEnqueuedOffset,
                         partitionProperties.LastEnqueuedTime,
                         _timeProvider.GetUtcNow());
+#pragma warning restore CS0618 // Type or member is obsolete
                 }
             }
         }
diff --git a/src/Spotflow.InMemory.Azure.EventHubs/Internals/InMemoryPartition.cs b/src/Spotflow.InMemory.Azure.EventHubs/Internals/InMemoryPartition.cs
@@ -72,6 +72,8 @@ public PartitionProperties GetProperties()
 
         var beginningSequenceNumber = currentSegment.Count > 0 ? currentSegment[0].SequenceNumber : lastSequenceNumber;
 
+#pragma warning disable CS0618 // Type or member is obsolete
+
         return EventHubsModelFactory.PartitionProperties(
               eventHubName: name,
               partitionId: PartitionId,
@@ -80,6 +82,8 @@ public PartitionProperties GetProperties()
               lastSequenceNumber: _lastSequenceNumber,
               lastOffset: _lastOffset,
               lastEnqueuedTime: _lastEnqueuedTime);
+
+#pragma warning restore CS0618 // Type or member is obsolete
     }
 
     public void SendEvent(EventData eventData, string? partitionKey)
@@ -114,6 +118,7 @@ public void SendEvent(EventData eventData, string? partitionKey)
             var eventDataPropertiesCopy = new Dictionary<string, object>(eventData.Properties);
             var eventDataSystemPropertiesCopy = new Dictionary<string, object>(eventData.SystemProperties);
 
+#pragma warning disable CS0618 // Type or member is obsolete
             var eventWithSystemProperties = EventHubsModelFactory.EventData(
                 eventBody: new(eventBodyCopy),
                 properties: eventDataPropertiesCopy,
@@ -123,6 +128,7 @@ public void SendEvent(EventData eventData, string? partitionKey)
                 offset: _lastOffset,
                 enqueuedTime: _lastEnqueuedTime
                );
+#pragma warning restore CS0618 // Type or member is obsolete
 
             eventWithSystemProperties.MessageId = eventData.MessageId;
             eventWithSystemProperties.CorrelationId = eventData.CorrelationId;
diff --git a/src/Spotflow.InMemory.Azure.EventHubs/Internals/StartingPosition.cs b/src/Spotflow.InMemory.Azure.EventHubs/Internals/StartingPosition.cs
@@ -45,9 +45,15 @@ public static InMemoryEventPosition FromEventPosition(EventPosition position)
 
         var isInclusive = ReflectionUtils.ReadInternalValueProperty<bool>(position, "IsInclusive");
 
-        var offset = ReflectionUtils.ReadInternalReferenceProperty<object>(position, "Offset");
+        if (ReflectionUtils.TryReadOptionalInternalReferenceProperty<object>(position, "OffsetString", out var offsetString) &&
+            offsetString is not null)
+        {
+            throw new NotSupportedException("EventPosition with offset is not supported.");
+        }
 
-        if (offset is not null)
+        // Offset was renamed to OffsetString after EventHub version 5.12.0, so we check for the old property name as well.
+        if (ReflectionUtils.TryReadOptionalInternalReferenceProperty<object>(position, "Offset", out var offset) &&
+            offset is not null)
         {
             throw new NotSupportedException("EventPosition with offset is not supported.");
         }
diff --git a/src/Spotflow.InMemory.Azure/Internals/ReflectionUtils.cs b/src/Spotflow.InMemory.Azure/Internals/ReflectionUtils.cs
@@ -4,28 +4,45 @@ namespace Spotflow.InMemory.Azure.Internals;
 
 internal static class ReflectionUtils
 {
-
     public static TOut? ReadInternalReferenceProperty<TOut>(object obj, string propertyName) where TOut : class
     {
-        var value = ReadInternalProperty(obj, propertyName);
+        if (!TryReadOptionalInternalReferenceProperty(obj, propertyName, out TOut? value))
+        {
+            throw new InvalidOperationException($"Property '{propertyName}' not found on type {obj.GetType().Name}");
+        }
+
+        return value;
+    }
+
+    public static bool TryReadOptionalInternalReferenceProperty<TOut>(object obj, string propertyName, out TOut? value) where TOut : class
+    {
+        value = null;
 
-        if (value is null)
+        if (!TryReadOptionalInternalProperty(obj, propertyName, out var rawValue))
         {
-            return null;
+            return false;
         }
 
-        if (!value.GetType().IsAssignableTo(typeof(TOut)))
+        if (rawValue is null)
         {
-            throw new InvalidOperationException($"Property '{propertyName}' is not assignable to {typeof(TOut).Name}");
+            return true; // null is a valid value for reference types
         }
 
-        return (TOut?) value;
+        if (!rawValue.GetType().IsAssignableTo(typeof(TOut)))
+        {
+            throw new InvalidOperationException($"Property '{propertyName}' is not assignable to {typeof(TOut).Name}");
+        }
 
+        value = (TOut?) rawValue;
+        return true;
     }
 
     public static TOut ReadInternalValueProperty<TOut>(object obj, string propertyName) where TOut : struct
     {
-        var value = ReadInternalProperty(obj, propertyName);
+        if (!TryReadOptionalInternalProperty(obj, propertyName, out var value))
+        {
+            throw new InvalidOperationException($"Property '{propertyName}' not found on type {obj.GetType().Name}");
+        }
 
         if (value is not TOut)
         {
@@ -35,7 +52,7 @@ public static TOut ReadInternalValueProperty<TOut>(object obj, string propertyNa
         return (TOut) value;
     }
 
-    private static object? ReadInternalProperty(object obj, string propertyName)
+    private static bool TryReadOptionalInternalProperty(object obj, string propertyName, out object? value)
     {
         ArgumentNullException.ThrowIfNull(obj);
         ArgumentException.ThrowIfNullOrWhiteSpace(propertyName);
@@ -46,10 +63,12 @@ public static TOut ReadInternalValueProperty<TOut>(object obj, string propertyNa
 
         if (property is null)
         {
-            throw new InvalidOperationException($"Property '{propertyName}' not found on type {obj.GetType().Name}");
+            value = null;
+            return false;
         }
 
-        return property.GetValue(obj);
+        value = property.GetValue(obj);
+        return true;
     }
 
 }
diff --git a/tests/Tests/EventHub/EventHubConsumerClientTests.cs b/tests/Tests/EventHub/EventHubConsumerClientTests.cs
@@ -55,17 +55,23 @@ public async Task GetPartitionProperties_Should_Return_Correct_Offset()
 
         var propertiesBeforeSend = await consumerClient.GetPartitionPropertiesAsync("0");
 
+#pragma warning disable CS0618 // Type or member is obsolete
         propertiesBeforeSend.LastEnqueuedOffset.Should().Be(-1);
+        propertiesBeforeSend.LastEnqueuedOffsetString.Should().Be("-1");
 
         await producerClient.SendAsync([new EventData()], new SendEventOptions { PartitionId = "0" });
 
         var propertiesAfterSend1 = await consumerClient.GetPartitionPropertiesAsync("0");
         propertiesAfterSend1.LastEnqueuedOffset.Should().Be(0);
+        propertiesAfterSend1.LastEnqueuedOffsetString.Should().Be("0");
 
         await producerClient.SendAsync([new EventData()], new SendEventOptions { PartitionId = "0" });
 
         var propertiesAfterSend2 = await consumerClient.GetPartitionPropertiesAsync("0");
         propertiesAfterSend2.LastEnqueuedOffset.Should().Be(26);
+        propertiesAfterSend2.LastEnqueuedOffsetString.Should().Be("26");
+
+#pragma warning restore CS0618 // Type or member is obsolete
 
     }
 }
diff --git a/tests/Tests/EventHub/EventHubProducerClientTests.cs b/tests/Tests/EventHub/EventHubProducerClientTests.cs
@@ -65,15 +65,23 @@ public async Task Offset_Should_Start_At_Zero_And_Increase_With_Each_Sent_Event(
 
         await producer.SendAsync([emptyEvent]);
         var emptyEventBatch = await consumer.ReceiveBatchAsync(1, TimeSpan.Zero);
+
+#pragma warning disable CS0618 // Type or member is obsolete
+
         emptyEventBatch.Single().Offset.Should().Be(0);
+        emptyEventBatch.Single().OffsetString.Should().Be("0");
 
         await producer.SendAsync([emptyEvent], new SendEventOptions { PartitionKey = "test-pk" });
         var emptyEventWithPartitionKey = await consumer.ReceiveBatchAsync(1, TimeSpan.Zero);
         emptyEventWithPartitionKey.Single().Offset.Should().Be(26);
+        emptyEventWithPartitionKey.Single().OffsetString.Should().Be("26");
 
         await producer.SendAsync([eventWithBody]);
         var eventWithBodyBatch = await consumer.ReceiveBatchAsync(1, TimeSpan.Zero);
         eventWithBodyBatch.Single().Offset.Should().Be(59);
+        eventWithBodyBatch.Single().OffsetString.Should().Be("59");
+
+#pragma warning restore CS0618 // Type or member is obsolete
     }
 
     [TestMethod]
diff --git a/tests/Tests/EventHub/EventHubTests.cs b/tests/Tests/EventHub/EventHubTests.cs
@@ -61,18 +61,25 @@ public async Task Custom_Partition_Initial_State_Should_Be_Used_For_All_Partitio
         batch0.Should().HaveCount(100);
         batch1.Should().HaveCount(100);
 
+#pragma warning disable CS0618 // Type or member is obsolete
+
         batch0.ElementAt(0).SequenceNumber.Should().Be(43);
         batch0.ElementAt(0).Offset.Should().Be(43);
+        batch0.ElementAt(0).OffsetString.Should().Be("43");
         batch0.ElementAt(0).EnqueuedTime.Should().Be(timeProvider.GetUtcNow());
         batch0.ElementAt(1).SequenceNumber.Should().Be(44);
         batch0.ElementAt(1).Offset.Should().Be(72);
+        batch0.ElementAt(1).OffsetString.Should().Be("72");
 
         batch1.ElementAt(0).SequenceNumber.Should().Be(43);
         batch1.ElementAt(0).Offset.Should().Be(43);
+        batch1.ElementAt(0).OffsetString.Should().Be("43");
         batch1.ElementAt(0).EnqueuedTime.Should().Be(timeProvider.GetUtcNow());
         batch1.ElementAt(1).SequenceNumber.Should().Be(44);
         batch1.ElementAt(1).Offset.Should().Be(72);
+        batch1.ElementAt(1).OffsetString.Should().Be("72");
 
+#pragma warning restore CS0618 // Type or member is obsolete
     }
 
     [TestMethod]
@@ -98,15 +105,19 @@ public async Task Default_Partition_Initial_State_Should_Be_Used_For_All_Partiti
         batch0.Should().HaveCount(1);
         batch1.Should().HaveCount(1);
 
-
+#pragma warning disable CS0618 // Type or member is obsolete
         batch0.Single().SequenceNumber.Should().Be(0);
         batch0.Single().Offset.Should().Be(0);
+        batch0.Single().OffsetString.Should().Be("0");
         batch0.Single().EnqueuedTime.Should().BeCloseTo(DateTimeOffset.UtcNow, TimeSpan.FromMinutes(2));
 
         batch1.Single().SequenceNumber.Should().Be(0);
         batch1.Single().Offset.Should().Be(0);
+        batch1.Single().OffsetString.Should().Be("0");
         batch1.Single().EnqueuedTime.Should().BeCloseTo(DateTimeOffset.UtcNow, TimeSpan.FromMinutes(2));
 
+#pragma warning restore CS0618 // Type or member is obsolete
+
     }
 
 }
diff --git a/tests/Tests/EventHub/PartitionReceiverTests.cs b/tests/Tests/EventHub/PartitionReceiverTests.cs
@@ -235,7 +235,11 @@ public async Task Last_Enqueued_Event_Properties_Should_Be_Refreshed()
 
         var properties1 = receiver.ReadLastEnqueuedEventProperties();
         properties1.SequenceNumber.Should().Be(0);
+
+#pragma warning disable CS0618 // Type or member is obsolete
+
         properties1.Offset.Should().Be(0);
+        properties1.OffsetString.Should().Be("0");
 
 
         // Send second event
@@ -246,6 +250,7 @@ public async Task Last_Enqueued_Event_Properties_Should_Be_Refreshed()
         var properties2 = receiver.ReadLastEnqueuedEventProperties();
         properties2.SequenceNumber.Should().Be(0);
         properties2.Offset.Should().Be(0);
+        properties2.OffsetString.Should().Be("0");
 
         _ = await receiver.ReceiveBatchAsync(100, TimeSpan.Zero);
 
@@ -255,6 +260,9 @@ public async Task Last_Enqueued_Event_Properties_Should_Be_Refreshed()
 
         properties3.SequenceNumber.Should().Be(1);
         properties3.Offset.Should().Be(37);
+        properties3.OffsetString.Should().Be("37");
+
+#pragma warning restore CS0618 // Type or member is obsolete
 
     }
 
diff --git a/tests/Tests/Storage/Blobs/BlobClientTests.cs b/tests/Tests/Storage/Blobs/BlobClientTests.cs
@@ -163,7 +163,7 @@ public void OpenWrite_And_Dispose_Immediately_Should_Create_Empty_Blob(BlobClien
             // Intentionally empty
         }
 
-        blobClient.DownloadContent().Value.Content.ToString().Should().BeEmpty();
+        blobClient.DownloadContent().Value.Content.ToArray().Should().BeEmpty();
 
         ShouldHaveBlocks(containerClient.GetBlockBlobClient(blobName), commited: 0, uncommited: 0);
 
@@ -185,7 +185,7 @@ public void OpenWrite_And_Without_Dispose_Should_Create_Empty_Blob(BlobClientTyp
 
         using var stream = OpenWrite(blobClient, true);
 
-        blobClient.DownloadContent().Value.Content.ToString().Should().BeEmpty();
+        blobClient.DownloadContent().Value.Content.ToArray().Should().BeEmpty();
 
         ShouldHaveBlocks(containerClient.GetBlockBlobClient(blobName), commited: 0, uncommited: 0);
     }

-Original file line number
+Diff line change
 # Solution-specific settings/overrides ------------------------------------------------------
 -## TBD
+-
+-
Original file line number	Diff line number	Diff line change
`@@ -218,11 +218,13 @@ private async Task<IEnumerable<EventData>> ReceiveBatchCoreAsync(int maximumEven`
`218`	`218`	`}`
`219`	`219`	`else`
`220`	`220`	`{`
	`221`	`+#pragma warning disable CS0618 // Type or member is obsolete`
`221`	`222`	`_lastEnqueuedEventProperties = new(`
`222`	`223`	`partitionProperties.LastEnqueuedSequenceNumber,`
`223`		`- partitionProperties.LastEnqueuedOffset,`
	`224`	`+ lastOffset: partitionProperties.LastEnqueuedOffset,`
`224`	`225`	`partitionProperties.LastEnqueuedTime,`
`225`	`226`	`_timeProvider.GetUtcNow());`
	`227`	`+#pragma warning restore CS0618 // Type or member is obsolete`
`226`	`228`	`}`
`227`	`229`	`}`
`228`	`230`	`}`
Original file line number	Diff line number	Diff line change
`@@ -4,28 +4,45 @@ namespace Spotflow.InMemory.Azure.Internals;`
`4`	`4`
`5`	`5`	`internal static class ReflectionUtils`
`6`	`6`	`{`
`7`		`-`
`8`	`7`	`public static TOut? ReadInternalReferenceProperty<TOut>(object obj, string propertyName) where TOut : class`
`9`	`8`	`{`
`10`		`- var value = ReadInternalProperty(obj, propertyName);`
	`9`	`+ if (!TryReadOptionalInternalReferenceProperty(obj, propertyName, out TOut? value))`
	`10`	`+ {`
	`11`	`+ throw new InvalidOperationException($"Property '{propertyName}' not found on type {obj.GetType().Name}");`
	`12`	`+ }`
	`13`	`+`
	`14`	`+ return value;`
	`15`	`+ }`
	`16`	`+`
	`17`	`+ public static bool TryReadOptionalInternalReferenceProperty<TOut>(object obj, string propertyName, out TOut? value) where TOut : class`
	`18`	`+ {`
	`19`	`+ value = null;`
`11`	`20`
`12`		`- if (value is null)`
	`21`	`+ if (!TryReadOptionalInternalProperty(obj, propertyName, out var rawValue))`
`13`	`22`	`{`
`14`		`- return null;`
	`23`	`+ return false;`
`15`	`24`	`}`
`16`	`25`
`17`		`- if (!value.GetType().IsAssignableTo(typeof(TOut)))`
	`26`	`+ if (rawValue is null)`
`18`	`27`	`{`
`19`		`- throw new InvalidOperationException($"Property '{propertyName}' is not assignable to {typeof(TOut).Name}");`
	`28`	`+ return true; // null is a valid value for reference types`
`20`	`29`	`}`
`21`	`30`
`22`		`- return (TOut?) value;`
	`31`	`+ if (!rawValue.GetType().IsAssignableTo(typeof(TOut)))`
	`32`	`+ {`
	`33`	`+ throw new InvalidOperationException($"Property '{propertyName}' is not assignable to {typeof(TOut).Name}");`
	`34`	`+ }`
`23`	`35`
	`36`	`+ value = (TOut?) rawValue;`
	`37`	`+ return true;`
`24`	`38`	`}`
`25`	`39`
`26`	`40`	`public static TOut ReadInternalValueProperty<TOut>(object obj, string propertyName) where TOut : struct`
`27`	`41`	`{`
`28`		`- var value = ReadInternalProperty(obj, propertyName);`
	`42`	`+ if (!TryReadOptionalInternalProperty(obj, propertyName, out var value))`
	`43`	`+ {`
	`44`	`+ throw new InvalidOperationException($"Property '{propertyName}' not found on type {obj.GetType().Name}");`
	`45`	`+ }`
`29`	`46`
`30`	`47`	`if (value is not TOut)`
`31`	`48`	`{`
`@@ -35,7 +52,7 @@ public static TOut ReadInternalValueProperty<TOut>(object obj, string propertyNa`
`35`	`52`	`return (TOut) value;`
`36`	`53`	`}`
`37`	`54`
`38`		`- private static object? ReadInternalProperty(object obj, string propertyName)`
	`55`	`+ private static bool TryReadOptionalInternalProperty(object obj, string propertyName, out object? value)`
`39`	`56`	`{`
`40`	`57`	`ArgumentNullException.ThrowIfNull(obj);`
`41`	`58`	`ArgumentException.ThrowIfNullOrWhiteSpace(propertyName);`
`@@ -46,10 +63,12 @@ public static TOut ReadInternalValueProperty<TOut>(object obj, string propertyNa`
`46`	`63`
`47`	`64`	`if (property is null)`
`48`	`65`	`{`
`49`		`- throw new InvalidOperationException($"Property '{propertyName}' not found on type {obj.GetType().Name}");`
	`66`	`+ value = null;`
	`67`	`+ return false;`
`50`	`68`	`}`
`51`	`69`
`52`		`- return property.GetValue(obj);`
	`70`	`+ value = property.GetValue(obj);`
	`71`	`+ return true;`
`53`	`72`	`}`
`54`	`73`
`55`	`74`	`}`