[Event Hubs Trigger] Work around service contract (Azure#50645)

* [Event Hubs Trigger] Work around service contract

The focus of these changes is to work around the
service contract for `offset` which is a string
that previously was always numeric.  In support
of the GeoDR feature, the service no longer
returns a numeric offset in all scenarios, sometimes
returning a format that cannot be parsed as a number.

To avoid applications that made assumptions about
the format, the `Offset` binding contracts for the
trigger were updated to always return a numeric
value as the string.  When the service returns an
offset that cannot be parsed as a number, the value
`-1` is substituted.  If used for readers, this would
possibly result in reading at the beginning of the
stream, causing duplicate processing of events but
will ensure there is no data loss.  Because the
`Offset` binding is not longer guaranteed to represent
the actual offset, a new `OffsetString` binding was
introduced to surface the actual offset value.

Author: jsquire

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/CHANGELOG.md

@@ -1,14 +1,15 @@
 # Release History
 
-## 6.5.2 (2025-06-12)
+## 6.5.2 (2025-06-16)
 
 ### Bugs Fixed
 
 - Fixed a bug where the data types of broker-owned properties were being adjusted when an event was read by the client, causing the underlying AMQP data to be mutated.  This resulted in binary changes when the AMQP message was serialized and unintentionally altered the service contract.  Going forward, the original data types will be preserved on the AMQP representation of the message and type normalization only applied to the .NET `EventData` projection.
-
 ### Other Changes
 
 - Updated the `Microsoft.Azure.Amqp` dependency to 2.7.0, which contains several bug fixes and adds support for AOT. _(see: [commits](https://github.com/Azure/azure-amqp/commits/hotfix/))_
+ 
+- Added a compatibility hack to the binding data associated with the trigger to preserve the previous service behavior of `Offset` being a valid numeric value.  As this data is not local to the trigger context and is surfaced upstream, the change in service behavior for GeoDR namespaces caused issues with some custom language workers making invalid assumptions about the string-based offset format.  To work around this, the `Offset` is being populated with `OffsetString` if it can be parsed as a number.  In the case that `OffsetString` is not a valid numeric value, it will default to `-1` which would position readers at the beginning of the stream, causing duplicates but preventing data loss.  A key for `OffsetString` was added that reflects the actual offset returned by the service in all scenarios.
 
 ## 6.5.1 (2025-04-09)
 

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Triggers/EventHubTriggerBindingStrategy.cs

@@ -54,6 +54,7 @@ public Dictionary<string, Type> GetBindingContract(bool isSingleDispatch = true)
 
             AddBindingContractMember(contract, "PartitionKey", typeof(string), isSingleDispatch);
             AddBindingContractMember(contract, "Offset", typeof(string), isSingleDispatch);
+            AddBindingContractMember(contract, "OffsetString", typeof(string), isSingleDispatch);
             AddBindingContractMember(contract, "SequenceNumber", typeof(long), isSingleDispatch);
             AddBindingContractMember(contract, "EnqueuedTimeUtc", typeof(DateTime), isSingleDispatch);
             AddBindingContractMember(contract, "Properties", typeof(IDictionary<string, object>), isSingleDispatch);
@@ -96,25 +97,36 @@ public Dictionary<string, object> GetBindingData(EventHubTriggerInput value)
 
         internal static void AddBindingData(Dictionary<string, object> bindingData, EventData[] events)
         {
-            int length = events.Length;
+            var length = events.Length;
             var partitionKeys = new string[length];
             var offsets = new string[length];
+            var offsetStrings = new string[length];
             var sequenceNumbers = new long[length];
             var enqueuedTimesUtc = new DateTime[length];
             var properties = new IDictionary<string, object>[length];
             var systemProperties = new IDictionary<string, object>[length];
 
             SafeAddValue(() => bindingData.Add("PartitionKeyArray", partitionKeys));
             SafeAddValue(() => bindingData.Add("OffsetArray", offsets));
+            SafeAddValue(() => bindingData.Add("OffsetStringArray", offsetStrings));
             SafeAddValue(() => bindingData.Add("SequenceNumberArray", sequenceNumbers));
             SafeAddValue(() => bindingData.Add("EnqueuedTimeUtcArray", enqueuedTimesUtc));
             SafeAddValue(() => bindingData.Add("PropertiesArray", properties));
             SafeAddValue(() => bindingData.Add("SystemPropertiesArray", systemProperties));
 
             for (int i = 0; i < events.Length; i++)
             {
+                if (!long.TryParse(events[i].OffsetString, NumberStyles.Integer, CultureInfo.InvariantCulture, out var offset))
+                {
+                    // Default to "beginning of stream" if parsing fails.  This will result in duplicates,
+                    // but ensures no data loss.
+
+                    offset = -1;
+                }
+
                 partitionKeys[i] = events[i].PartitionKey;
-                offsets[i] = events[i].OffsetString;
+                offsets[i] = offset.ToString(CultureInfo.InvariantCulture);
+                offsetStrings[i] = events[i].OffsetString;
                 sequenceNumbers[i] = events[i].SequenceNumber;
                 enqueuedTimesUtc[i] = events[i].EnqueuedTime.DateTime;
                 properties[i] = events[i].Properties;
@@ -124,8 +136,17 @@ internal static void AddBindingData(Dictionary<string, object> bindingData, Even
 
         private static void AddBindingData(Dictionary<string, object> bindingData, EventData eventData)
         {
+            if (!long.TryParse(eventData.OffsetString, NumberStyles.Integer, CultureInfo.InvariantCulture, out var offset))
+            {
+                // Default to "beginning of stream" if parsing fails.  This will result in duplicates,
+                // but ensures no data loss.
+
+                offset = -1;
+            }
+
             SafeAddValue(() => bindingData.Add("PartitionKey", eventData.PartitionKey));
-            SafeAddValue(() => bindingData.Add("Offset", eventData.OffsetString));
+            SafeAddValue(() => bindingData.Add("Offset", offset.ToString(CultureInfo.InvariantCulture)));
+            SafeAddValue(() => bindingData.Add("OffsetString", eventData.OffsetString));
             SafeAddValue(() => bindingData.Add("SequenceNumber", eventData.SequenceNumber));
             SafeAddValue(() => bindingData.Add("EnqueuedTimeUtc", eventData.EnqueuedTime.DateTime));
             SafeAddValue(() => bindingData.Add("Properties", eventData.Properties));
@@ -154,8 +175,18 @@ private static IDictionary<string, object> GetSystemPropertiesForBinding(EventDa
             }
 
             // Following is needed to maintain structure of bindingdata: https://github.com/Azure/azure-webjobs-sdk/pull/1849
+
+            if (!long.TryParse(eventData.OffsetString, NumberStyles.Integer, CultureInfo.InvariantCulture, out var offset))
+            {
+                // Default to "beginning of stream" if parsing fails.  This will result in duplicates,
+                // but ensures no data loss.
+
+                offset = -1;
+            }
+
             modifiedDictionary["SequenceNumber"] = eventData.SequenceNumber;
-            modifiedDictionary["Offset"] = eventData.OffsetString;
+            modifiedDictionary["Offset"] = offset;
+            modifiedDictionary["OffsetString"] = eventData.OffsetString;
             modifiedDictionary["PartitionKey"] = eventData.PartitionKey;
             modifiedDictionary["EnqueuedTimeUtc"] = eventData.EnqueuedTime.DateTime;
             return modifiedDictionary;

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/src/Triggers/EventHubTriggerInput.cs

@@ -63,10 +63,19 @@ public Dictionary<string, string> GetTriggerDetails(EventProcessorPartition cont
                 return new Dictionary<string, string>();
             }
 
-            string offset, enqueueTimeUtc, sequenceNumber;
+            string offset, offsetString, enqueueTimeUtc, sequenceNumber;
             if (IsSingleDispatch)
             {
-                offset = Events[0].OffsetString;
+                if (!long.TryParse(Events[0].OffsetString, NumberStyles.Integer, CultureInfo.InvariantCulture, out var offsetLong))
+                {
+                    // Default to "beginning of stream" if parsing fails.  This will result in duplicates,
+                    // but ensures no data loss.
+
+                    offsetLong = -1;
+                }
+
+                offset = offsetLong.ToString(CultureInfo.InvariantCulture);
+                offsetString = Events[0].OffsetString;
                 enqueueTimeUtc = Events[0].EnqueuedTime.ToString("o", CultureInfo.InvariantCulture);
                 sequenceNumber = Events[0].SequenceNumber.ToString(CultureInfo.InvariantCulture);
             }
@@ -76,18 +85,26 @@ public Dictionary<string, string> GetTriggerDetails(EventProcessorPartition cont
                 EventData last = Events[Events.Length - 1];
 
                 offset = $"{first.OffsetString}-{last.OffsetString}";
+                offsetString = $"{first.OffsetString}-{last.OffsetString}";
                 enqueueTimeUtc = $"{first.EnqueuedTime.ToString("o", CultureInfo.InvariantCulture)}-{last.EnqueuedTime.ToString("o", CultureInfo.InvariantCulture)}";
                 sequenceNumber = $"{first.SequenceNumber}-{last.SequenceNumber}";
             }
 
             return new Dictionary<string, string>()
             {
                 { "PartitionId", context.PartitionId },
-                { "Offset", offset },
+                { "OffsetString", offsetString },
                 { "EnqueueTimeUtc", enqueueTimeUtc },
                 { "SequenceNumber", sequenceNumber },
                 { "Count", Events.Length.ToString(CultureInfo.InvariantCulture)},
 
+                // Preserve a numeric offset for compatibility with existing code.  It is immportant
+                // to note that this does not conform to the Event Hubs service contract and may not
+                // reflect the actual offset of the event in the stream.  In the case that the
+                // offset is not a valid numeric value, it will default to -1 which would position
+                // readers at the beginning of the stream, causing duplicates but preventing data loss.
+                { "Offset", offset },
+
                 // Preserve a misspelling that existed in the original code, as
                 // there may be applications relying on this.
                 { "PartionId", context.PartitionId }

sdk/eventhub/Microsoft.Azure.WebJobs.Extensions.EventHubs/tests/EventHubTests.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 using System.Text;
 using Azure.Messaging.EventHubs;
 using Azure.Messaging.EventHubs.Consumer;
@@ -26,10 +27,11 @@ public void GetStaticBindingContract_ReturnsExpectedValue()
             var strategy = new EventHubTriggerBindingStrategy();
             var contract = strategy.GetBindingContract();
 
-            Assert.AreEqual(8, contract.Count);
+            Assert.AreEqual(9, contract.Count);
             Assert.AreEqual(typeof(TriggerPartitionContext), contract["TriggerPartitionContext"]);
             Assert.AreEqual(typeof(PartitionContext), contract["PartitionContext"]);
             Assert.AreEqual(typeof(string), contract["Offset"]);
+            Assert.AreEqual(typeof(string), contract["OffsetString"]);
             Assert.AreEqual(typeof(long), contract["SequenceNumber"]);
             Assert.AreEqual(typeof(DateTime), contract["EnqueuedTimeUtc"]);
             Assert.AreEqual(typeof(IDictionary<string, object>), contract["Properties"]);
@@ -42,10 +44,11 @@ public void GetBindingContract_SingleDispatch_ReturnsExpectedValue()
             var strategy = new EventHubTriggerBindingStrategy();
             var contract = strategy.GetBindingContract(true);
 
-            Assert.AreEqual(8, contract.Count);
+            Assert.AreEqual(9, contract.Count);
             Assert.AreEqual(typeof(TriggerPartitionContext), contract["TriggerPartitionContext"]);
             Assert.AreEqual(typeof(PartitionContext), contract["PartitionContext"]);
             Assert.AreEqual(typeof(string), contract["Offset"]);
+            Assert.AreEqual(typeof(string), contract["OffsetString"]);
             Assert.AreEqual(typeof(long), contract["SequenceNumber"]);
             Assert.AreEqual(typeof(DateTime), contract["EnqueuedTimeUtc"]);
             Assert.AreEqual(typeof(IDictionary<string, object>), contract["Properties"]);
@@ -58,11 +61,12 @@ public void GetBindingContract_MultipleDispatch_ReturnsExpectedValue()
             var strategy = new EventHubTriggerBindingStrategy();
             var contract = strategy.GetBindingContract(false);
 
-            Assert.AreEqual(8, contract.Count);
+            Assert.AreEqual(9, contract.Count);
             Assert.AreEqual(typeof(TriggerPartitionContext), contract["TriggerPartitionContext"]);
             Assert.AreEqual(typeof(PartitionContext), contract["PartitionContext"]);
             Assert.AreEqual(typeof(string[]), contract["PartitionKeyArray"]);
             Assert.AreEqual(typeof(string[]), contract["OffsetArray"]);
+            Assert.AreEqual(typeof(string[]), contract["OffsetStringArray"]);
             Assert.AreEqual(typeof(long[]), contract["SequenceNumberArray"]);
             Assert.AreEqual(typeof(DateTime[]), contract["EnqueuedTimeUtcArray"]);
             Assert.AreEqual(typeof(IDictionary<string, object>[]), contract["PropertiesArray"]);
@@ -80,18 +84,24 @@ public void GetBindingData_SingleDispatch_ReturnsExpectedValue()
             var strategy = new EventHubTriggerBindingStrategy();
             var bindingData = strategy.GetBindingData(input);
 
-            Assert.AreEqual(8, bindingData.Count);
+            if (!long.TryParse(evt.OffsetString, NumberStyles.Integer, CultureInfo.InvariantCulture, out long offsetLong))
+            {
+                offsetLong = -1;
+            }
+
+            Assert.AreEqual(9, bindingData.Count);
             Assert.AreSame(input.ProcessorPartition.PartitionContext, bindingData["TriggerPartitionContext"]);
             Assert.AreSame(input.ProcessorPartition.PartitionContext, bindingData["PartitionContext"]);
             Assert.AreEqual(evt.PartitionKey, bindingData["PartitionKey"]);
-            Assert.AreEqual(evt.OffsetString, bindingData["Offset"]);
+            Assert.AreEqual(offsetLong.ToString(CultureInfo.InvariantCulture), bindingData["Offset"]);
+            Assert.AreEqual(evt.OffsetString, bindingData["OffsetString"]);
             Assert.AreEqual(evt.SequenceNumber, bindingData["SequenceNumber"]);
             Assert.AreEqual(evt.EnqueuedTime.DateTime, bindingData["EnqueuedTimeUtc"]);
             Assert.AreSame(evt.Properties, bindingData["Properties"]);
             IDictionary<string, object> bindingDataSysProps = bindingData["SystemProperties"] as Dictionary<string, object>;
             Assert.NotNull(bindingDataSysProps);
             Assert.AreEqual(bindingDataSysProps["PartitionKey"], bindingData["PartitionKey"]);
-            Assert.AreEqual(bindingDataSysProps["Offset"], bindingData["Offset"]);
+            Assert.AreEqual(offsetLong.ToString(CultureInfo.InvariantCulture), bindingData["Offset"]);
             Assert.AreEqual(bindingDataSysProps["SequenceNumber"], bindingData["SequenceNumber"]);
             Assert.AreEqual(bindingDataSysProps["EnqueuedTimeUtc"], bindingData["EnqueuedTimeUtc"]);
             Assert.AreEqual(bindingDataSysProps["iothub-connection-device-id"], "testDeviceId");
@@ -123,13 +133,14 @@ public void GetBindingData_MultipleDispatch_ReturnsExpectedValue()
             var strategy = new EventHubTriggerBindingStrategy();
             var bindingData = strategy.GetBindingData(input);
 
-            Assert.AreEqual(8, bindingData.Count);
+            Assert.AreEqual(9, bindingData.Count);
             Assert.AreSame(input.ProcessorPartition.PartitionContext, bindingData["TriggerPartitionContext"]);
             Assert.AreSame(input.ProcessorPartition.PartitionContext, bindingData["PartitionContext"]);
 
             // verify an array was created for each binding data type
             Assert.AreEqual(events.Length, ((string[])bindingData["PartitionKeyArray"]).Length);
             Assert.AreEqual(events.Length, ((string[])bindingData["OffsetArray"]).Length);
+            Assert.AreEqual(events.Length, ((string[])bindingData["OffsetStringArray"]).Length);
             Assert.AreEqual(events.Length, ((long[])bindingData["SequenceNumberArray"]).Length);
             Assert.AreEqual(events.Length, ((DateTime[])bindingData["EnqueuedTimeUtcArray"]).Length);
             Assert.AreEqual(events.Length, ((IDictionary<string, object>[])bindingData["PropertiesArray"]).Length);