Fixes #6675: Update IEventHubDataAdapter to support StreamId to partition mapping (#6676) (#6731)

* Fixes #6675: Include stream Namespace in PartitionKey for Event Hub messages

* Update IEventHubDataAdapter to support partition to StreamId mapping

Co-authored-by: Alex Meyer-Gleaves <[email protected]>

Author: benjaminpetit

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/EventHubAdapterFactory.cs

@@ -225,7 +225,8 @@ public virtual Task QueueMessageBatchAsync<T>(Guid streamGuid, string streamName
             Dictionary<string, object> requestContext)
         {
             EventData eventData = this.dataAdapter.ToQueueMessage(streamGuid, streamNamespace, events, token, requestContext);
-            return this.client.SendAsync(eventData, streamGuid.ToString());
+            string partitionKey = this.dataAdapter.GetPartitionKey(streamGuid, streamNamespace);
+            return this.client.SendAsync(eventData, partitionKey);
         }
 
         /// <summary>
@@ -321,4 +322,4 @@ public static EventHubAdapterFactory Create(IServiceProvider services, string na
             return factory;
         }
     }
-}
+}

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/EventHubDataAdapter.cs

@@ -76,13 +76,10 @@ public virtual CachedMessage FromQueueMessage(StreamPosition streamPosition, Eve
 
         public virtual StreamPosition GetStreamPosition(string partition, EventData queueMessage)
         {
-            Guid streamGuid =
-            Guid.Parse(queueMessage.SystemProperties.PartitionKey);
-            string streamNamespace = queueMessage.GetStreamNamespaceProperty();
-            IStreamIdentity stremIdentity = new StreamIdentity(streamGuid, streamNamespace);
+            IStreamIdentity streamIdentity = this.GetStreamIdentity(queueMessage);
             StreamSequenceToken token =
                 new EventHubSequenceTokenV2(queueMessage.SystemProperties.Offset, queueMessage.SystemProperties.SequenceNumber, 0);
-            return new StreamPosition(stremIdentity, token);
+            return new StreamPosition(streamIdentity, token);
         }
 
         /// <summary>
@@ -95,6 +92,26 @@ public virtual string GetOffset(CachedMessage lastItemPurged)
             return SegmentBuilder.ReadNextString(lastItemPurged.Segment, ref readOffset); // read offset
         }
 
+        /// <summary>
+        /// Get the Event Hub partition key to use for a stream.
+        /// </summary>
+        /// <param name="streamGuid">The stream Guid.</param>
+        /// <param name="streamNamespace">The stream Namespace.</param>
+        /// <returns>The partition key to use for the stream.</returns>
+        public virtual string GetPartitionKey(Guid streamGuid, string streamNamespace) => streamGuid.ToString();
+
+        /// <summary>
+        /// Get the <see cref="IStreamIdentity"/> for an event message.
+        /// </summary>
+        /// <param name="queueMessage">The event message.</param>
+        /// <returns>The stream identity.</returns>
+        public virtual IStreamIdentity GetStreamIdentity(EventData queueMessage)
+        {
+            Guid streamGuid = Guid.Parse(queueMessage.SystemProperties.PartitionKey);
+            string streamNamespace = queueMessage.GetStreamNamespaceProperty();
+            return new StreamIdentity(streamGuid, streamNamespace);
+        }
+
         // Placed object message payload into a segment.
         protected virtual ArraySegment<byte> EncodeMessageIntoSegment(EventData queueMessage, Func<int, ArraySegment<byte>> getSegment)
         {

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/IEventHubDataAdapter.cs

@@ -12,5 +12,9 @@ public interface IEventHubDataAdapter : IQueueDataAdapter<EventData>, ICacheData
         StreamPosition GetStreamPosition(string partition, EventData queueMessage);
 
         string GetOffset(CachedMessage cachedMessage);
+
+        string GetPartitionKey(Guid streamGuid, string streamNamespace);
+
+        IStreamIdentity GetStreamIdentity(EventData queueMessage);
     }
 }