feat(eventhubs): add buffered producer client

[Johnathan W (j7nw4r)]

Summary

azure_messaging_eventhubs publishes only explicit batches today. The caller builds a batch and sends it. This change adds BufferedProducerClient, which accepts single events and publishes them in the background. The client does the bounded buffering, the partition assignment, the batching, the back pressure, the retries, the delivery reports, the flush, and the shutdown. .NET, Java, JavaScript, and Python have an equivalent client. Rust and Go do not.

Motivation

An application that publishes many events must build the batches itself, or pay one round trip for each event. An application that builds the batches also owns the partition assignment, the batch size limits, the retry handling, and the shutdown order. Each of these is easy to get wrong, and each of these is already in this crate.

Changes

• Added BufferedProducerClient and BufferedProducerClientBuilder, with enqueue_event, enqueue_events, flush, close, abort, total_buffered_event_count, and buffered_event_count. The builder has open and open_with_connection_string.
• Added EnqueueEventOptions, SendBatchSucceededContext, SendBatchFailedContext, and ErrorKind::SendNotAccepted.
• Moved the serialization state of EventDataBatch into an owned pub(crate) EventDataBatchInner, so a background worker can hold a batch without a borrow of the ProducerClient. EventDataBatch<'a> keeps its lifetime and calls the new type, so every public signature stays the same.
• Added ProducerClient::send_batch_envelope, which returns the AMQP outcome directly. send_batch now calls it and keeps its current behavior.
• Depends on the claims-based-security fix of fix(eventhubs): serialize claims-based-security authorizations #4895, which is now merged. The service permits one $cbs link for each connection, and two authorizations that overlapped made the service reject the second one with NotAllowed. The live tests here found it: this client starts one sender for each partition, and 7 of 32 events failed. This branch carried that fix while it was in review. The rebase onto main drops the duplicate commit, so the change comes from main now.
• Added an example program, a README section, a changelog entry, and the live tests.

The change adds no dependency. The workers start through azure_core::async_runtime, and the partition buffers use an async-lock semaphore. There is no production dependency on tokio, and no second retry, connection, or task-lifecycle system: the workers publish through the current RecoverableSender.

Delivery semantics

• A successful enqueue means only that the local buffer accepted the event. It does not mean that Event Hubs accepted the event.
• Each terminal send attempt gives exactly one success report or one failure report, which carries the events and the partition ID.
• Modified and Released are failures. Neither outcome shows that the service kept the events, so the buffered path reports both with ErrorKind::SendNotAccepted. ProducerClient::send_batch still maps them to Ok(()), so this change is additive.
• The client reports a failure only after the retry policy is fully used, or when the error is not retryable. It never enqueues a terminally failed batch again.
• flush puts a barrier in the queue of each partition. An event enqueued behind the barrier does not delay the flush.
• close sends the buffered events and then releases the resources. abort discards them. A drop stops the background work and logs a warning.

Points for the API review

• The delivery reports use registered async handlers, and the failure handler is required. The four sibling SDKs do the same. A Stream of results was not selected, because a slow or absent consumer either grows the memory without a bound or stops the workers.
• The client has close() and abort(), and not close(bool). Both take &self and are idempotent, so an application can keep the client in an Arc. ProducerClient::close(self) is different.
• enqueue_event returns Result<()>, and not the buffered count that .NET and JavaScript return. A count from an enqueue can look like a delivery signal, and total_buffered_event_count() already gives it.
• EnqueueEventOptions is not #[non_exhaustive], which matches SendEventOptions and EventDataBatchOptions. The two result contexts are #[non_exhaustive].
• The defaults are a 1 second maximum wait time and 1500 buffered events for each partition, which match .NET, JavaScript, and Python. The Java builder gives 30 seconds and is the only different one.
• The partition assignment obeys .NET PartitionResolver: round robin, and a Jenkins lookup3 hash of the partition key. The tests hold the hash port to the 12 values from the .NET test suite, so a key reaches the same partition as in the other SDKs.
• The send concurrency for one partition is always 1, so the events keep their enqueue order. There is no option for it.

Validation

cargo fmt --check, cargo clippy --all-targets --all-features -- -D warnings, cargo test --all-features (186 unit tests and 54 doc tests, no failures), and RUSTDOCFLAGS='-Dwarnings' cargo doc all pass. The Build Analyze gate also passes locally.

39 of the unit tests cover the buffered producer. They run through a crate-internal BufferedSendClient seam with a mock that gives prepared AMQP outcomes and per-call gates, so they need no network and no test uses a sleep to show a race. The full set ran 20 times in sequence with no failure.

The live tests pass against a real Event Hub with 5 partitions: the enqueue and receive round trip, explicit partition routing, the partial-batch timeout, the graceful close, automatic partition assignment, the connection-string client, and the in-crate forced-error recovery test, which makes sure that a link failure loses no event and reports no event two times. The five Microsoft Entra ID tests ran through a temporary local change that authenticates with a connection string, because the test identity holds no Send role on that namespace. Their committed form needs a run by someone whose identity has the role.

Remaining gates

• The API review. This change adds a new public client surface. The APIView artifact regenerates with cargo run --manifest-path eng/tools/generate_api_report/Cargo.toml -- --package azure_messaging_eventhubs; the repository .gitignore excludes review/, so this branch does not contain it.
• The partition list refresh. The client reads the partition IDs one time when it opens. .NET reads them again every 10 seconds, so a partition added later goes unused until the client restarts. This affects the throughput and not the correctness. The rustdoc gives the limitation, and a subsequent change can add the refresh.
• The transport type, if feat(eventhubs): support AMQP-over-WebSockets transport #4596 merges. feat(eventhubs): support AMQP-over-WebSockets transport #4596 adds with_transport_type to the producer and the consumer builder. BufferedProducerClientBuilder forwards application_id, retry_options, and custom_endpoint to ProducerClient::builder(), and it needs the same forward for the transport type. Without that one method, a buffered producer cannot use AMQP over WebSockets. Neither change blocks the other, and the second one to merge adds the method.
• Related work that does not block this change. fix(eventhubs): discard stale resources across connection recovery #4571, fix(eventhubs): abort Authorizer token-refresh task on connection tea… #4598, fix(eventhubs): honor EventDataBatchOptions::max_size_in_bytes #4808, and feat(eventhubs): support AMQP-over-WebSockets transport #4596 are open, and none is a functional prerequisite. fix(eventhubs): honor EventDataBatchOptions::max_size_in_bytes #4808 edits producer/batch.rs, and feat(eventhubs): support AMQP-over-WebSockets transport #4596 edits producer/mod.rs and the changelog, which this change also edits. The second change to merge needs a small rebase.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
2 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

Pull request overview

Adds a buffered Event Hubs producer for bounded, asynchronous batching and delivery reporting.

Changes:

• Adds the buffered producer public API, routing options, delivery contexts, and SendNotAccepted.
• Implements partition workers, batching, backpressure, flushing, shutdown, and routing.
• Adds CBS authorization serialization, documentation, examples, and tests.

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
tests/eventhubs_buffered_producer.rs	Adds live buffered-producer tests.
src/producer/mod.rs	Adds raw batch-envelope sending.
src/producer/buffered/worker.rs	Implements partition workers.
src/producer/buffered/tests.rs	Adds buffered-producer unit tests.
src/producer/buffered/send_client.rs	Abstracts production and mock sending.
src/producer/buffered/partition_resolver.rs	Implements partition assignment.
src/producer/buffered/mod.rs	Defines the client, builder, and lifecycle.
src/producer/batch.rs	Extracts owned batch serialization state.
src/lib.rs	Exports the new public API.
src/error.rs	Adds SendNotAccepted.
src/common/recoverable/connection.rs	Adds CBS serialization locking.
src/common/recoverable/claims_based_security.rs	Applies the CBS lock.
README.md	Documents buffered publishing.
examples/eventhubs_buffered_produce_events.rs	Adds a usage example.
CHANGELOG.md	Records the feature and CBS fix.

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/worker.rs:423

• These permits are released only after the success/failure handler has finished. If this batch owns every permit for the partition, a handler that enqueues to the same partition deadlocks: its enqueue waits for capacity that this worker releases only after that handler returns. Release the counters and permits as soon as the send reaches its terminal outcome, before awaiting the delivery handler.

        self.release(event_count);
        // The events reached a terminal outcome, so the buffer has space again.
        drop(permits);

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/mod.rs:632

• A concurrent second close or abort returns Ok here as soon as the first shutdown sets closed, even though workers may still be sending and the connection is not yet released. This contradicts the shutdown methods' wait guarantees; a close/abort race also silently ignores the second caller's requested mode. Coordinate shutdown through shared completion state so every caller waits for the in-progress shutdown (and define close-versus-abort precedence).

    async fn shutdown(&self, abandon: bool) -> Result<()> {
        if self.closed.swap(true, Ordering::AcqRel) {
            return Ok(());

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md:7

• This changelog entry is not concise: it combines the public API summary with detailed batching behavior, defaults, cross-SDK comparisons, and shutdown semantics. The changelog guideline requires each public API change to be summarized briefly on one line; keep the detailed behavior in the README and API docs.

- 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))

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

Suppressed comments (3)

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/mod.rs:632

• closed records that shutdown started, not that it completed, but every concurrent or later caller returns Ok immediately. Thus a second close()/abort() can report success while the first call is still sending and invoking handlers; if the first future is cancelled or its producer close fails, no later call can finish or observe that failure. Use a shared shutdown completion/result (or an async lifecycle state) so all callers await the in-progress shutdown rather than treating it as complete.

        if self.closed.swap(true, Ordering::AcqRel) {
            return Ok(());

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/mod.rs:514

• An enqueue can clone this sender, increment both counters, and then race with abort(): Tokio may drop the receiver, while shutdown resets total_buffered to zero at line 687. If unbounded_send then fails, this rollback subtracts from zero and wraps the total to usize::MAX. Keep the sender lock held through the synchronous accounting and publish so shutdown cannot reset the counters until either the command is queued or the rollback is complete.

        // Take a clone of the sender and release the lock at once, so a slow
        // enqueue never blocks a close.
        let sender = {
            let guard = state.sender.lock().unwrap();
            guard.as_ref().ok_or_else(Self::closed_error)?.clone()
        };
        // Count the event before the worker can see it. The worker decrements
        // the counts as soon as the event reaches a terminal outcome, and a
        // fast terminal path (an oversized event with a handler that returns at
        // once) can run before this call returns. Counting afterwards lets that
        // decrement reach zero first and wrap the counts to `usize::MAX`.
        state.buffered.fetch_add(1, Ordering::AcqRel);
        self.total_buffered.fetch_add(1, Ordering::AcqRel);

        if sender.unbounded_send(command).is_err() {
            // The worker never saw the event, so it never decrements for it.
            state.buffered.fetch_sub(1, Ordering::AcqRel);
            self.total_buffered.fetch_sub(1, Ordering::AcqRel);
            return Err(Self::closed_error());
        }

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/worker.rs:211

• The acknowledgement is sent before PartitionWorker is dropped, so on the std-thread runtime abort() can receive it and return while this worker still owns send_client (and therefore the producer connection) and the delivery handlers. This contradicts the shutdown guarantee at mod.rs:618-620 and explains the Arc::try_unwrap failure path there. Drop the worker-owned fields before signaling completion.

        // Tell the client that this worker holds nothing more. The client waits
        // for this, so a close does not return while a worker still runs.
        if let Some(stopped) = self.stopped.take() {
            let _ = stopped.send(());
        }

[Copilot]

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

Suppressed comments (6)

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/mod.rs:674

• On the standard runtime, task.abort() only detaches the worker. If abort() starts while a worker is awaiting a user delivery handler, this acknowledgement cannot resolve until that arbitrary future completes and may hang forever; the retry policy does not bound handler duration. Make handler execution cancellation-aware (for example, select it against an explicit worker cancellation signal) so abort() can actually complete independently of a stuck handler.

        // Wait for every worker to stop. A cancelled task drops its end of the
        // channel, which resolves the receiver with an error, so this waits for
        // the worker to finish or to be dropped, and never for both.
        for acknowledgement in acknowledgements {
            let _ = acknowledgement.await;

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/mod.rs:633

• This marks the client closed before shutdown completes, so a concurrent second close or abort returns Ok(()) immediately while the first call may still be sending, invoking handlers, and holding the connection. That violates both methods' completion guarantees. Calls that observe an in-progress shutdown should await a shared shutdown-completion result; only calls after completion should return immediately.

        if self.closed.swap(true, Ordering::AcqRel) {
            return Ok(());
        }

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/mod.rs:308

• Without the tokio feature, this runtime call creates a dedicated std::thread (sdk/core/typespec_client_core/src/async_runtime/standard_runtime.rs:106-127). Because the loop eagerly starts one worker per partition, even idle clients consume one long-lived OS thread per partition, which does not scale to high-partition-count hubs. Please multiplex partition workers on a bounded/shared executor or start a bounded number lazily.

            let task = get_async_runtime().spawn(Box::pin(worker.run()));

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/worker.rs:254

• On the supported non-Tokio runtime, polling azure_core::sleep spawns a new OS thread that sleeps for the full duration (sdk/core/typespec_client_core/src/async_runtime/standard_runtime.rs:163-205). This timer is created for every new batch; when a batch fills early, dropping the future does not stop its sleeper thread. High-throughput configurations such as a buffer count of 1 can therefore create an unbounded burst of concurrent threads. Please use a shared/cancelable timer mechanism rather than creating one runtime sleep per batch.

                if timer.is_none() {
                    *timer = Some(Box::pin(sleep(self.max_wait_time)));
                }

sdk/eventhubs/azure_messaging_eventhubs/src/producer/mod.rs:473

• get_sender eagerly runs ensure_sender before entering RecoverableSender::send, whose retry loop performs the same ensure. A retryable link-attach error can therefore return here immediately, contradicting this helper's guarantee that Err means retries were exhausted or the error was non-retryable. Construct the recoverable sender directly so acquisition happens inside its retry loop.

        let sender = self.connection.get_sender(path).await?;

sdk/eventhubs/azure_messaging_eventhubs/src/producer/buffered/send_client.rs:61

• This sender setup runs outside the send retry loop, and the worker turns any error from max_message_size into an immediate terminal failure for an already accepted event. A transient attach/recovery error can therefore drain the buffered queue as failures without applying the configured retry policy. Resolve the link maximum through the same retry/recovery path before reporting the event as terminally failed.

    async fn max_message_size(&self, partition_id: &str) -> Result<u64> {
        let path = self.partition_path(partition_id)?;
        let sender = self.producer.ensure_sender(path).await?;
        sender.max_message_size().await?.ok_or_else(|| {