feat(eventhubs): add buffered producer client

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- Added `BufferedProducerClient`, a producer that accepts single events and publishes them in the background. One worker owns each partition, so events keep their enqueued order and a slow partition does not stop the others. A successful enqueue means only that the local buffer accepted the event; the client reports each terminal outcome through the handlers that `with_on_send_succeeded` and `with_on_send_failed` register, and a handler for failed batches is required. The client sends a batch when the next event does not fit, when the batch reaches the configured event count, when the maximum wait time expires, on `flush`, and on `close`. The defaults are a 1 second maximum wait time and 1500 buffered events for each partition, matching the .NET, JavaScript, and Python clients. `flush` sets a barrier over the events accepted before the call, `close` sends the buffered events, and `abort` abandons them. New public types: `BufferedProducerClient`, `BufferedProducerClientBuilder`, `EnqueueEventOptions`, `SendBatchSucceededContext`, and `SendBatchFailedContext`. ([#4873](https://github.com/Azure/azure-sdk-for-rust/pull/4873))
+- Added the `ErrorKind::SendNotAccepted` error variant. The buffered producer reports an AMQP `Modified` or `Released` outcome as a delivery failure, because neither outcome means that the service durably stored the events. `ProducerClient::send_batch` keeps its historical behavior and treats both as success with a warning.
 - Added connection-string authentication. `ProducerClientBuilder` and `ConsumerClientBuilder` now have an `open_with_connection_string` method that authenticates with a Shared Access Signature parsed from an Event Hubs connection string (`Endpoint=sb://...;SharedAccessKeyName=...;SharedAccessKey=...`, optionally with `EntityPath`, or a pre-formed `SharedAccessSignature`). The connection-string parser is exposed publicly as `ConnectionString`. This reaches parity with the other Azure SDKs for development and test scenarios; Microsoft Entra ID via `open` with a `TokenCredential` remains the recommended path for production. The parser rejects empty required values and empty Event Hub names up front, and a pre-formed `SharedAccessSignature` reports its own `se` as the token expiry (rather than a rolling client-side window); because such a token cannot be renewed, the connection's token refresher detects the non-advancing expiry and leaves the broker to enforce it. ([#3459](https://github.com/Azure/azure-sdk-for-rust/issues/3459))
 - The `EventProcessor` now opens every partition receiver with AMQP epoch (owner level) `0` and surfaces broker-initiated displacement as the new `EventHubsError::ConsumerDisconnected` error kind. When a second `EventProcessor` instance claims a partition this instance is currently holding, the broker disconnects this instance's receiver and the consumer's `stream_events()` resolves with `ConsumerDisconnected`. This matches the behavior of `EventProcessorClient` in the .NET and Java Azure SDKs. Consumers should pattern-match on `ErrorKind::ConsumerDisconnected` to detect a stolen partition and re-acquire a client via `next_partition_client()`.
 - Added `EventHubsError::ConsumerDisconnected(Option<AmqpDescribedError>)` error variant.

sdk/eventhubs/azure_messaging_eventhubs/README.md

@@ -120,6 +120,10 @@ Additional examples for various scenarios can be found on in the examples direct
 - [Send events](#send-events)
   - [Send events directly to the Event Hub](#send-events-directly-to-the-event-hub)
   - [Send events using a batch operation](#send-events-using-a-batch-operation)
+- [Send events with the buffered producer](#send-events-with-the-buffered-producer)
+  - [Route events to a partition](#route-events-to-a-partition)
+  - [Flush and shut down](#flush-and-shut-down)
+  - [Trade-offs of buffered publishing](#trade-offs-of-buffered-publishing)
 - [Open an Event Hubs message consumer on an Event Hubs instance](#open-an-event-hubs-message-consumer-on-an-event-hub-instance)
 - [Receive events](#receive-events)
 
@@ -178,6 +182,125 @@ async fn send_events(producer: &ProducerClient) -> Result<(), Box<dyn std::error
 }
 ```
 
+### Send events with the buffered producer
+
+`BufferedProducerClient` accepts single events and publishes them in the background. The client
+groups the events into batches for each partition, and one worker for each partition sends them.
+This gives a higher throughput than `ProducerClient`, because the caller does not wait for each
+send.
+
+The client reports the outcome of each batch through handlers. A handler for failed batches is
+required, because a send failure arrives after the enqueue call already returned.
+
+```rust no_run
+use azure_identity::DeveloperToolsCredential;
+use azure_messaging_eventhubs::BufferedProducerClient;
+
+async fn buffered_publish() -> Result<(), Box<dyn std::error::Error>> {
+    let host = "<EVENTHUBS_HOST>";
+    let eventhub = "<EVENTHUB_NAME>";
+    let credential = DeveloperToolsCredential::new(None)?;
+
+    let producer = BufferedProducerClient::builder()
+        .with_on_send_succeeded(|context| async move {
+            println!(
+                "The service accepted {} events on partition {}.",
+                context.events.len(),
+                context.partition_id
+            );
+        })
+        .with_on_send_failed(|context| async move {
+            eprintln!(
+                "{} events failed on partition {}: {}",
+                context.events.len(),
+                context.partition_id,
+                context.error
+            );
+        })
+        .open(host, eventhub, credential.clone())
+        .await?;
+
+    for index in 0..1000 {
+        producer.enqueue_event(format!("event {index}"), None).await?;
+    }
+
+    producer.close().await?;
+    Ok(())
+}
+```
+
+#### Route events to a partition
+
+Give a partition ID to send an event to one partition. Give a partition key to send every event
+with that key to the same partition. Set at most one of the two; the client rejects a request that
+sets both. When you set neither, the client assigns the partitions in round-robin order.
+
+```rust no_run
+use azure_messaging_eventhubs::{BufferedProducerClient, EnqueueEventOptions};
+
+async fn route_events(
+    producer: &BufferedProducerClient,
+) -> Result<(), Box<dyn std::error::Error>> {
+    producer
+        .enqueue_event(
+            "to partition 0",
+            Some(EnqueueEventOptions {
+                partition_id: Some("0".to_string()),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    producer
+        .enqueue_event(
+            "grouped by key",
+            Some(EnqueueEventOptions {
+                partition_key: Some("customer-17".to_string()),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    Ok(())
+}
+```
+
+#### Flush and shut down
+
+`flush` sets a barrier. It completes once every event that the client accepted before the call
+reaches a terminal outcome. An event that arrives after the barrier does not delay the call.
+
+`close` sends the buffered events and then shuts the client down. `abort` shuts the client down at
+once and abandons the buffered events.
+
+```rust no_run
+use azure_messaging_eventhubs::BufferedProducerClient;
+
+async fn flush_and_close(
+    producer: &BufferedProducerClient,
+) -> Result<(), Box<dyn std::error::Error>> {
+    producer.enqueue_event("an event", None).await?;
+
+    // Wait for the events that the client already accepted.
+    producer.flush().await?;
+    println!("{} events are still buffered.", producer.total_buffered_event_count());
+
+    // Send what is left, then shut down.
+    producer.close().await?;
+    Ok(())
+}
+```
+
+#### Trade-offs of buffered publishing
+
+- A successful enqueue means only that the local buffer accepted the event. It does not mean that
+  Event Hubs accepted the event.
+- The process loses the buffered events if it stops before a flush or a close. Call `flush` or
+  `close` when the delivery of the buffered events matters.
+- A send failure arrives after the enqueue call already returned, through the failure handler.
+- Buffering gives a higher throughput, but the latency of one event is less predictable.
+- Use `ProducerClient` when the application needs the result of each send.
+
 ### Open an Event Hubs message consumer on an Event Hub instance
 
 ```rust no_run

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_buffered_produce_events.rs

@@ -0,0 +1,124 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+// cspell: ignore retryable
+
+//! This sample shows how to publish events with the buffered producer client.
+//!
+//! The caller enqueues single events. The client buffers them, groups them into
+//! batches for each partition, and publishes them in the background.
+//!
+//! A successful enqueue means only that the local buffer accepted the event. It
+//! does not mean that Event Hubs accepted the event. The client reports the real
+//! outcome through the two handlers below.
+
+use azure_core::time::Duration;
+use azure_identity::DeveloperToolsCredential;
+use azure_messaging_eventhubs::{BufferedProducerClient, EnqueueEventOptions};
+use std::sync::{
+    atomic::{AtomicUsize, Ordering},
+    Arc,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let eventhub_namespace = std::env::var("EVENTHUBS_HOST")?;
+    let eventhub_name = std::env::var("EVENTHUB_NAME")?;
+    let credential = DeveloperToolsCredential::new(None)?;
+
+    // The handlers report the outcome of each batch. Count the events so the
+    // sample can print a summary at the end.
+    let accepted = Arc::new(AtomicUsize::new(0));
+    let rejected = Arc::new(AtomicUsize::new(0));
+
+    let for_success = accepted.clone();
+    let for_failure = rejected.clone();
+
+    let producer = BufferedProducerClient::builder()
+        // Send a batch that is not full after this time.
+        .with_max_wait_time(Duration::seconds(1))
+        // Wait for space once a partition buffer holds this many events.
+        .with_max_buffered_event_count_per_partition(1500)
+        .with_on_send_succeeded(move |context| {
+            let accepted = for_success.clone();
+            async move {
+                accepted.fetch_add(context.events.len(), Ordering::AcqRel);
+                println!(
+                    "The service accepted {} events on partition {}.",
+                    context.events.len(),
+                    context.partition_id
+                );
+            }
+        })
+        // A handler for failed batches is required. The client calls it only
+        // after the retry policy is exhausted, or when the error is not
+        // retryable. The client does not enqueue the events again, so the
+        // application decides what to do with them.
+        .with_on_send_failed(move |context| {
+            let rejected = for_failure.clone();
+            async move {
+                rejected.fetch_add(context.events.len(), Ordering::AcqRel);
+                eprintln!(
+                    "{} events failed on partition {}: {}",
+                    context.events.len(),
+                    context.partition_id,
+                    context.error
+                );
+            }
+        })
+        .open(
+            eventhub_namespace.as_str(),
+            eventhub_name.as_str(),
+            credential.clone(),
+        )
+        .await?;
+
+    // The client assigns these events to the partitions in round-robin order.
+    for index in 0..100 {
+        producer
+            .enqueue_event(format!("automatic event {index}"), None)
+            .await?;
+    }
+
+    // These events all go to partition 0.
+    producer
+        .enqueue_events(
+            vec!["first", "second", "third"],
+            Some(EnqueueEventOptions {
+                partition_id: Some("0".to_string()),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    // Every event with the same key goes to the same partition.
+    producer
+        .enqueue_event(
+            "an event for one customer",
+            Some(EnqueueEventOptions {
+                partition_key: Some("customer-17".to_string()),
+                ..Default::default()
+            }),
+        )
+        .await?;
+
+    println!(
+        "{} events are waiting in the buffer.",
+        producer.total_buffered_event_count()
+    );
+
+    // The flush completes once every event that the client accepted before this
+    // call reaches a terminal outcome.
+    producer.flush().await?;
+
+    // A graceful close sends what is left, then it releases the connection. Use
+    // `abort` instead to shut down at once and abandon the buffered events.
+    producer.close().await?;
+
+    println!(
+        "Done. The service accepted {} events and rejected {} events.",
+        accepted.load(Ordering::Acquire),
+        rejected.load(Ordering::Acquire)
+    );
+    Ok(())
+}

sdk/eventhubs/azure_messaging_eventhubs/src/error.rs

@@ -45,6 +45,17 @@ pub enum ErrorKind {
     ///
     AmqpError(AmqpError),
 
+    /// The service settled the transfer, but it did not durably accept it.
+    ///
+    /// The broker returned an AMQP `Modified` or `Released` outcome. Neither
+    /// outcome means the service stored the events. The buffered producer
+    /// reports this as a delivery failure.
+    ///
+    /// A `Released` or `Modified` outcome does not prove that the service
+    /// discarded the events either. If the caller sends the same events again,
+    /// the service can store them two times.
+    SendNotAccepted(Cow<'static, str>),
+
     /// Receiver was disconnected by the broker because another receiver
     /// attached with the same or higher epoch (owner level). The inner
     /// `AmqpDescribedError` is for logging; match on the variant:
@@ -95,6 +106,9 @@ impl std::fmt::Display for EventHubsError {
             ErrorKind::SendRejected(e) => write!(f, "Send rejected: {:?}", e),
             ErrorKind::InvalidManagementResponse => f.write_str("Invalid management response"),
             ErrorKind::AmqpError(source) => write!(f, "AMQP Error: {:?}", source),
+            ErrorKind::SendNotAccepted(msg) => {
+                write!(f, "Send was not durably accepted: {}", msg)
+            }
             ErrorKind::ConsumerDisconnected(e) => {
                 write!(
                     f,

sdk/eventhubs/azure_messaging_eventhubs/src/lib.rs

@@ -23,6 +23,10 @@ pub use consumer::{
 };
 pub use producer::{
     batch::{EventDataBatch, EventDataBatchOptions},
+    buffered::{
+        BufferedProducerClient, EnqueueEventOptions, SendBatchFailedContext,
+        SendBatchSucceededContext,
+    },
     ProducerClient, SendBatchOptions, SendEventOptions, SendMessageOptions,
 };
 
@@ -36,6 +40,7 @@ pub use event_processor::{processor::EventProcessor, CheckpointStore, ProcessorS
 pub mod builders {
     pub use crate::consumer::builders::ConsumerClientBuilder;
     pub use crate::event_processor::processor::builders::EventProcessorBuilder;
+    pub use crate::producer::buffered::builders::BufferedProducerClientBuilder;
     pub use crate::producer::builders::ProducerClientBuilder;
 }
 pub use common::connection_string::ConnectionString;