[Event Hubs Client] Publish Events Without Batch (#11651)

The focus of these changes is to allow the publishing of events without the
overhead of creating an explicit batch.  In this scenario the caller assumes
the risk with respect to exceeding the maximum size allowed by the Event Hubs
service, trading safety for lower overhead and higher throughput.

Author: jsquire

sdk/eventhub/Azure.Messaging.EventHubs/api/Azure.Messaging.EventHubs.netstandard2.0.cs

@@ -409,12 +409,10 @@ public enum ProcessingStoppedReason
 }
 namespace Azure.Messaging.EventHubs.Producer
 {
-    public partial class CreateBatchOptions
+    public partial class CreateBatchOptions : Azure.Messaging.EventHubs.Producer.SendEventOptions
     {
         public CreateBatchOptions() { }
         public long? MaximumSizeInBytes { get { throw null; } set { } }
-        public string PartitionId { get { throw null; } set { } }
-        public string PartitionKey { get { throw null; } set { } }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
@@ -455,6 +453,8 @@ public EventHubProducerClient(string connectionString, string eventHubName, Azur
         public virtual System.Threading.Tasks.Task<string[]> GetPartitionIdsAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task<Azure.Messaging.EventHubs.PartitionProperties> GetPartitionPropertiesAsync(string partitionId, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task SendAsync(Azure.Messaging.EventHubs.Producer.EventDataBatch eventBatch, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task SendAsync(System.Collections.Generic.IEnumerable<Azure.Messaging.EventHubs.EventData> eventBatch, Azure.Messaging.EventHubs.Producer.SendEventOptions options, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public virtual System.Threading.Tasks.Task SendAsync(System.Collections.Generic.IEnumerable<Azure.Messaging.EventHubs.EventData> eventBatch, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override string ToString() { throw null; }
     }
@@ -470,6 +470,18 @@ public EventHubProducerClientOptions() { }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override string ToString() { throw null; }
     }
+    public partial class SendEventOptions
+    {
+        public SendEventOptions() { }
+        public string PartitionId { get { throw null; } set { } }
+        public string PartitionKey { get { throw null; } set { } }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public override bool Equals(object obj) { throw null; }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public override int GetHashCode() { throw null; }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public override string ToString() { throw null; }
+    }
 }
 namespace Microsoft.Extensions.Azure
 {

sdk/eventhub/Azure.Messaging.EventHubs/src/Producer/CreateBatchOptions.cs

@@ -11,7 +11,7 @@ namespace Azure.Messaging.EventHubs.Producer
     ///   behaves and is sent to the Event Hubs service.
     /// </summary>
     ///
-    public class CreateBatchOptions
+    public class CreateBatchOptions : SendEventOptions
     {
         /// <summary>The requested maximum size to allow for the batch, in bytes.</summary>
         private long? _maximumSizeInBytes = null;
@@ -40,58 +40,6 @@ public long? MaximumSizeInBytes
             }
         }
 
-        /// <summary>
-        ///   Allows a hashing key to be provided for the batch of events, which instructs the Event Hubs
-        ///   service map this key to a specific partition but allowing the service to choose an arbitrary,
-        ///   partition for this batch of events and any other batches using the same partition hashing key.
-        ///
-        ///   The selection of a partition is stable for a given partition hashing key.  Should any other
-        ///   batches of events be sent using the same exact partition hashing key, the Event Hubs service will
-        ///   route them all to the same partition.
-        ///
-        ///   This should be specified only when there is a need to group events by partition, but there is
-        ///   flexibility into which partition they are routed. If ensuring that a batch of events is sent
-        ///   only to a specific partition, it is recommended that the identifier of the position be
-        ///   specified directly when sending the batch.
-        /// </summary>
-        ///
-        /// <value>
-        ///   If the producer wishes to influence the automatic routing of events to partitions, the partition
-        ///   hashing key to associate with the event or batch of events; otherwise, <c>null</c>.
-        /// </value>
-        ///
-        /// <remarks>
-        ///   If the <see cref="CreateBatchOptions.PartitionKey" /> is specified, then no <see cref="CreateBatchOptions.PartitionId" />
-        ///   may be set when sending.
-        /// </remarks>
-        ///
-        public string PartitionKey { get; set; }
-
-        /// <summary>
-        ///   If specified, events will be published to this specific partition.  If the identifier is not
-        ///   specified, the Event Hubs service will be responsible for routing events automatically to an available partition.
-        /// </summary>
-        ///
-        /// <value>
-        ///   If the producer wishes the events to be automatically routed to partitions, <c>null</c>; otherwise, the identifier
-        ///   of the desired partition.
-        /// </value>
-        ///
-        /// <remarks>
-        ///   If the <see cref="CreateBatchOptions.PartitionId" /> is specified, then no <see cref="CreateBatchOptions.PartitionKey" />
-        ///   may be set when sending.
-        ///
-        ///   <para>Allowing automatic routing of partitions is recommended when:</para>
-        ///   <para>- The sending of events needs to be highly available.</para>
-        ///   <para>- The event data should be evenly distributed among all available partitions.</para>
-        ///
-        ///   If no partition is specified, the following rules are used for automatically selecting one:
-        ///   <para>1) Distribute the events equally amongst all available partitions using a round-robin approach.</para>
-        ///   <para>2) If a partition becomes unavailable, the Event Hubs service will automatically detect it and forward the message to another available partition.</para>
-        /// </remarks>
-        ///
-        public string PartitionId { get; set; }
-
         /// <summary>
         ///   Creates a new copy of the current <see cref="CreateBatchOptions" />, cloning its attributes into a new instance.
         /// </summary>
@@ -106,15 +54,6 @@ internal CreateBatchOptions Clone() =>
                 _maximumSizeInBytes = MaximumSizeInBytes
             };
 
-        /// <summary>
-        ///   Converts the <see cref="CreateBatchOptions" /> into an equivalent
-        ///   <see cref="SendEventOptions" /> instance.
-        /// </summary>
-        ///
-        /// <returns>A set of sending options equivalent to those represented by the batch options.</returns>
-        ///
-        internal SendEventOptions ToSendOptions() => new SendEventOptions(PartitionId, PartitionKey);
-
         /// <summary>
         ///   Determines whether the specified <see cref="System.Object" /> is equal to this instance.
         /// </summary>

sdk/eventhub/Azure.Messaging.EventHubs/src/Producer/EventHubProducerClient.cs

@@ -370,53 +370,64 @@ internal virtual async Task SendAsync(EventData eventData,
         }
 
         /// <summary>
-        ///   Sends a set of events to the associated Event Hub using a batched approach.  If the size of events exceed the
-        ///   maximum size of a single batch, an exception will be triggered and the send will fail.
+        ///   Sends a set of events to the associated Event Hub using a batched approach.  Because the batch is implicitly created, the size of the event set is not
+        ///   validated until this method is invoked.  The call will fail if the size of the specified set of events exceeds the maximum allowable size of a single batch.
         /// </summary>
         ///
-        /// <param name="events">The set of event data to send.</param>
+        /// <param name="eventBatch">The set of event data to send.</param>
         /// <param name="cancellationToken">An optional <see cref="CancellationToken" /> instance to signal the request to cancel the operation.</param>
         ///
         /// <returns>A task to be resolved on when the operation has completed.</returns>
         ///
-        /// <seealso cref="SendAsync(IEnumerable{EventData}, SendEventOptions, CancellationToken)"/>
+        /// <exception cref="EventHubsException">
+        ///   Occurs when the set of events exceeds the maximum size allowed in a single batch, as determined by the Event Hubs service.  The <see cref="EventHubsException.Reason" /> will be set to
+        ///   <see cref="EventHubsException.FailureReason.MessageSizeExceeded"/> in this case.
+        /// </exception>
+        ///
+        /// <seealso cref="SendAsync(IEnumerable{EventData}, SendEventOptions, CancellationToken)" />
+        /// <seealso cref="SendAsync(EventDataBatch, CancellationToken)" />
+        /// <seealso cref="CreateBatchAsync(CancellationToken)" />
         ///
-        internal virtual async Task SendAsync(IEnumerable<EventData> events,
-                                              CancellationToken cancellationToken = default) => await SendAsync(events, null, cancellationToken).ConfigureAwait(false);
+        public virtual async Task SendAsync(IEnumerable<EventData> eventBatch,
+                                            CancellationToken cancellationToken = default) => await SendAsync(eventBatch, null, cancellationToken).ConfigureAwait(false);
 
         /// <summary>
-        ///   Sends a set of events to the associated Event Hub using a batched approach.  If the size of events exceed the
-        ///   maximum size of a single batch, an exception will be triggered and the send will fail.
+        ///   Sends a set of events to the associated Event Hub using a batched approach.  Because the batch is implicitly created, the size of the event set is not
+        ///   validated until this method is invoked.  The call will fail if the size of the specified set of events exceeds the maximum allowable size of a single batch.
         /// </summary>
         ///
-        /// <param name="events">The set of event data to send.</param>
+        /// <param name="eventBatch">The set of event data to send.</param>
         /// <param name="options">The set of options to consider when sending this batch.</param>
         /// <param name="cancellationToken">An optional <see cref="CancellationToken" /> instance to signal the request to cancel the operation.</param>
         ///
         /// <returns>A task to be resolved on when the operation has completed.</returns>
         ///
-        /// <seealso cref="SendAsync(EventData, CancellationToken)" />
-        /// <seealso cref="SendAsync(EventData, SendEventOptions, CancellationToken)" />
+        /// <exception cref="EventHubsException">
+        ///   Occurs when the set of events exceeds the maximum size allowed in a single batch, as determined by the Event Hubs service.  The <see cref="EventHubsException.Reason" /> will be set to
+        ///   <see cref="EventHubsException.FailureReason.MessageSizeExceeded"/> in this case.
+        /// </exception>
+        ///
         /// <seealso cref="SendAsync(IEnumerable{EventData}, CancellationToken)" />
         /// <seealso cref="SendAsync(EventDataBatch, CancellationToken)" />
+        /// <seealso cref="CreateBatchAsync(CreateBatchOptions, CancellationToken)" />
         ///
-        internal virtual async Task SendAsync(IEnumerable<EventData> events,
-                                              SendEventOptions options,
-                                              CancellationToken cancellationToken = default)
+        public virtual async Task SendAsync(IEnumerable<EventData> eventBatch,
+                                            SendEventOptions options,
+                                            CancellationToken cancellationToken = default)
         {
             options ??= DefaultSendOptions;
 
-            Argument.AssertNotNull(events, nameof(events));
+            Argument.AssertNotNull(eventBatch, nameof(eventBatch));
             AssertSinglePartitionReference(options.PartitionId, options.PartitionKey);
 
             int attempts = 0;
 
-            events = (events as IList<EventData>) ?? events.ToList();
-            InstrumentMessages(events);
+            eventBatch = (eventBatch as IList<EventData>) ?? eventBatch.ToList();
+            InstrumentMessages(eventBatch);
 
             var diagnosticIdentifiers = new List<string>();
 
-            foreach (var eventData in events)
+            foreach (var eventData in eventBatch)
             {
                 if (EventDataInstrumentation.TryExtractDiagnosticId(eventData, out var identifier))
                 {
@@ -433,8 +444,7 @@ internal virtual async Task SendAsync(IEnumerable<EventData> events,
                 try
                 {
                     await using var _ = pooledProducer.ConfigureAwait(false);
-
-                    await pooledProducer.TransportProducer.SendAsync(events, options, cancellationToken).ConfigureAwait(false);
+                    await pooledProducer.TransportProducer.SendAsync(eventBatch, options, cancellationToken).ConfigureAwait(false);
 
                     return;
                 }
@@ -468,10 +478,6 @@ internal virtual async Task SendAsync(IEnumerable<EventData> events,
         ///
         /// <returns>A task to be resolved on when the operation has completed.</returns>
         ///
-        /// <seealso cref="SendAsync(EventData, CancellationToken)" />
-        /// <seealso cref="SendAsync(EventData, SendEventOptions, CancellationToken)" />
-        /// <seealso cref="SendAsync(IEnumerable{EventData}, CancellationToken)" />
-        /// <seealso cref="SendAsync(IEnumerable{EventData}, SendEventOptions, CancellationToken)" />
         /// <seealso cref="CreateBatchAsync(CancellationToken)" />
         ///
         public virtual async Task SendAsync(EventDataBatch eventBatch,
@@ -535,6 +541,7 @@ public virtual async Task SendAsync(EventDataBatch eventBatch,
         /// <returns>An <see cref="EventDataBatch" /> with the default batch options.</returns>
         ///
         /// <seealso cref="CreateBatchAsync(CreateBatchOptions, CancellationToken)" />
+        /// <seealso cref="SendAsync(EventDataBatch, CancellationToken)" />
         ///
         public virtual async ValueTask<EventDataBatch> CreateBatchAsync(CancellationToken cancellationToken = default) => await CreateBatchAsync(null, cancellationToken).ConfigureAwait(false);
 
@@ -552,7 +559,8 @@ public virtual async Task SendAsync(EventDataBatch eventBatch,
         ///
         /// <returns>An <see cref="EventDataBatch" /> with the requested <paramref name="options"/>.</returns>
         ///
-        /// <seealso cref="CreateBatchAsync(CreateBatchOptions, CancellationToken)" />
+        /// <seealso cref="CreateBatchAsync(CancellationToken)" />
+        /// <seealso cref="SendAsync(EventDataBatch, CancellationToken)" />
         ///
         public virtual async ValueTask<EventDataBatch> CreateBatchAsync(CreateBatchOptions options,
                                                                         CancellationToken cancellationToken = default)
@@ -561,7 +569,7 @@ public virtual async ValueTask<EventDataBatch> CreateBatchAsync(CreateBatchOptio
             AssertSinglePartitionReference(options.PartitionId, options.PartitionKey);
 
             TransportEventBatch transportBatch = await PartitionProducerPool.EventHubProducer.CreateBatchAsync(options, cancellationToken).ConfigureAwait(false);
-            return new EventDataBatch(transportBatch, FullyQualifiedNamespace, EventHubName, options.ToSendOptions());
+            return new EventDataBatch(transportBatch, FullyQualifiedNamespace, EventHubName, options);
         }
 
         /// <summary>

sdk/eventhub/Azure.Messaging.EventHubs/src/Producer/SendEventOptions.cs

@@ -7,10 +7,10 @@ namespace Azure.Messaging.EventHubs.Producer
 {
     /// <summary>
     ///   The set of options that can be specified to influence the way in which events
-    ///   are sent to the Event Hubs service.
+    ///   are published to the Event Hubs service.
     /// </summary>
     ///
-    internal class SendEventOptions
+    public class SendEventOptions
     {
         /// <summary>
         ///   Allows a hashing key to be provided for the batch of events, which instructs the Event Hubs