From f8103136285f50655662493441d73a0c602ccb05 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 14:24:41 -0400 Subject: [PATCH 1/2] test(eventhubs): pin the deferred receiver attach log `open_receiver_on_partition` builds an `EventReceiver` and returns. The AMQP link attaches on the first `stream_events()` poll, but the method logs "Receiver attached on partition." before it returns. The log claims an attach that did not happen, and it claims it again when the attach later fails. Add three tests in the `consumer::tests` module: - `open_receiver_on_partition_logs_a_deferred_attach` asserts the log says the link attaches on the first poll, and that the old attach claim is absent. This test is red until the source changes. - `open_receiver_on_partition_defers_the_attach_to_the_first_poll` arms an attach error and shows that the open succeeds and the first poll fails. It pins the lazy contract. - `open_receiver_on_partition_never_logs_an_attach_that_failed` shows that a failed attach records no attach message. Add a `LogBuffer` writer and a `capture_logs` helper that install a thread-local `tracing` subscriber, plus two helpers that build an unconnected client. The tests use `MockCredential` and reach no network. --- .../src/consumer/mod.rs | 143 +++++++++++++++++- 1 file changed, 139 insertions(+), 4 deletions(-) diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs index 205e7092e90..1e2252f2e84 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs @@ -883,15 +883,15 @@ pub mod builders { #[cfg(test)] pub(crate) mod tests { use crate::{ - common::tests::force_errors, models::EventData, ConsumerClient, EventDataBatchOptions, - ProducerClient, Result, StartLocation, StartPosition, + common::tests::force_errors, error::ErrorKind, models::EventData, ConsumerClient, + EventDataBatchOptions, ProducerClient, Result, StartLocation, StartPosition, }; use azure_core::{sleep::sleep, time::Duration}; use azure_core_amqp::{error::AmqpErrorKind, AmqpError, AmqpTransport}; - use azure_core_test::{recorded, TestContext}; + use azure_core_test::{credentials::MockCredential, recorded, TestContext}; use futures::stream::StreamExt; use std::{ - sync::Arc, + sync::{Arc, Mutex}, time::{SystemTime, UNIX_EPOCH}, }; @@ -1337,4 +1337,139 @@ pub(crate) mod tests { }) .await } + /// Collects the formatted output of a `tracing` subscriber, so a test can + /// read the records that the code under test made. + #[derive(Clone, Default)] + struct LogBuffer(Arc>>); + + impl LogBuffer { + fn contents(&self) -> String { + let buffer = self.0.lock().expect("the log buffer lock is poisoned"); + String::from_utf8_lossy(&buffer).to_string() + } + } + + impl std::io::Write for LogBuffer { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + let mut buffer = self.0.lock().expect("the log buffer lock is poisoned"); + buffer.extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for LogBuffer { + type Writer = LogBuffer; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + /// Installs a log capture and returns it with its guard. The dispatcher is + /// thread local, so the test must stay on one thread. Drop the guard + /// before you read `contents()`. + fn capture_logs() -> (LogBuffer, tracing::subscriber::DefaultGuard) { + let buffer = LogBuffer::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(buffer.clone()) + .with_max_level(tracing::Level::TRACE) + .with_ansi(false) + .finish(); + let guard = tracing::subscriber::set_default(subscriber); + (buffer, guard) + } + + fn unconnected_consumer() -> ConsumerClient { + ConsumerClient::new_unconnected( + "example.servicebus.windows.net", + "test-eventhub", + Arc::new(MockCredential), + ) + .expect("the client must build") + } + + fn consumer_with_armed_attach_error() -> ConsumerClient { + let consumer = unconnected_consumer(); + consumer + .recoverable_connection() + .force_attach_error(AmqpError::with_message("attach failed")) + .expect("the attach error must arm"); + consumer + } + + #[test] + fn open_receiver_on_partition_logs_a_deferred_attach() { + let consumer = unconnected_consumer(); + let (buffer, guard) = capture_logs(); + futures::executor::block_on(consumer.open_receiver_on_partition("0".to_string(), None)) + .expect("the open must succeed without a connection"); + drop(guard); + let logs = buffer.contents(); + + assert!( + logs.contains( + "Created receiver on partition. The AMQP link attaches on the first stream_events() poll." + ), + "the log must say the link attaches on the first poll, got: {logs}" + ); + assert!( + !logs.contains("Receiver attached on partition."), + "the log must not claim an attach happened, got: {logs}" + ); + } + + // Pins the lazy contract. An eager + // `recoverable_connection.get_receiver(...)` put before the + // `Ok(EventReceiver::new(...))` return makes the armed attach error leave + // the open, and this test fails. + #[tokio::test] + async fn open_receiver_on_partition_defers_the_attach_to_the_first_poll() { + let consumer = consumer_with_armed_attach_error(); + let receiver = consumer + .open_receiver_on_partition("0".to_string(), None) + .await + .expect("the open must not attach, so the armed attach error must not surface here"); + + let mut stream = std::pin::pin!(receiver.stream_events()); + let error = stream + .next() + .await + .expect("the stream yields the armed attach failure") + .expect_err("the armed attach error must surface on the first poll"); + assert!( + matches!(error.kind, ErrorKind::AmqpError(_)), + "expected AmqpError, got {:?}", + error.kind + ); + } + + #[tokio::test] + async fn open_receiver_on_partition_never_logs_an_attach_that_failed() { + let consumer = consumer_with_armed_attach_error(); + let (buffer, guard) = capture_logs(); + let receiver = consumer + .open_receiver_on_partition("0".to_string(), None) + .await + .expect("the open must succeed without a connection"); + + let mut stream = std::pin::pin!(receiver.stream_events()); + let _ = stream.next().await; + drop(guard); + let logs = buffer.contents(); + + // The first assertion anchors the capture. An empty capture would make + // the second assertion pass for the wrong reason. + assert!( + logs.contains("Opening receiver on partition."), + "the capture must hold the events of the call, got: {logs}" + ); + assert!( + !logs.contains("Attached receiver on partition."), + "no attach was made, so nothing may record one, got: {logs}" + ); + } } From 9a8deed2e38c0387d1c0bcb63295a03452b10442 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 14:31:06 -0400 Subject: [PATCH 2/2] fix(eventhubs): stop logging an attach that has not happened yet `ConsumerClient::open_receiver_on_partition` does no network I/O, but it wrote `info!("Receiver attached on partition.")` before it returned the `EventReceiver`. The AMQP link attaches on the first poll of `stream_events()`, and the attach closure in the recoverable connection already writes the truthful record. A reader of the log saw an attach that had not happened, and the log kept that claim even when the attach failed later. The message now says the call created the receiver, and that the link attaches on the first poll. The level, the position, and the four fields stay the same. No eager attach is added. The documentation on `open_receiver_on_partition`, `EventProcessor::run` and `PartitionClient::stream_events` now states where the attach happens, and that the service reports an unknown consumer group or an unknown partition id from that first poll. The old text also named `MessageReceiver`, a type that does not exist. Refs #5094 --- .../azure_messaging_eventhubs/CHANGELOG.md | 1 + .../azure_messaging_eventhubs/src/consumer/mod.rs | 14 ++++++++++---- .../src/event_processor/partition_client.rs | 4 ++++ .../src/event_processor/processor.rs | 8 +++++++- 4 files changed, 22 insertions(+), 5 deletions(-) diff --git a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md index 4696c059971..a954efdcc2f 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md +++ b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md @@ -32,6 +32,7 @@ - Fixed a deadlock when a CBS failure during management-client creation started connection recovery. ([#4728](https://github.com/Azure/azure-sdk-for-rust/issues/4728)) - Closed a stale-resource window in connection recovery. A `ReconnectConnection` recovery that fired while a slow-path attach (authorize, session begin, or sender/receiver link attach) was in flight could cache a resource bound to the just-dropped connection; the next operation on that resource failed (unauthorized / detached / closed) and triggered a second, redundant recovery cycle. A recovery generation counter now tags each cached resource, and a slow path that completes across a recovery discards its result and re-attaches against the new connection instead of caching the stale one. The authorizer's token cache is mutable (a background task refreshes tokens) so it cannot use the same one-shot cell as the connection caches; both of its writers, `authorize_path` and the refresh task, instead re-check the generation under the same lock that recovery's clear takes, and a recovery brackets its invalidation with a generation bump on each side, which leaves the counter odd for as long as the recovery runs, so a slow path that overlaps a recovery at either end also discards rather than caching a resource bound to the connection that recovery is dropping. A token refresh pass that a recovery discards now applies the same backoff floor as a failed pass, so a recovery storm cannot turn the refresh loop into an uncapped stream of credential and CBS calls. The per-path / per-partition concurrency is preserved. ([#4454](https://github.com/Azure/azure-sdk-for-rust/issues/4454)) - `InMemoryCheckpointStore` now rotates the ETag and refreshes `last_modified_time` when an existing ownership is renewed, matching the create path and the production `BlobCheckpointStore`. Previously the renewal path reinserted the caller's record verbatim, leaving a stale ETag and timestamp; that divergence from the real store could mask bugs in code that relies on ETag rotation for optimistic concurrency. ([#4594](https://github.com/Azure/azure-sdk-for-rust/issues/4594)) +- `ConsumerClient::open_receiver_on_partition` no longer logs `Receiver attached on partition.` before an attach happens. The call creates the receiver and does no network I/O. The AMQP link attaches on the first poll of `EventReceiver::stream_events()`, which is where the service reports an unknown consumer group or an unknown partition id. The documentation on `open_receiver_on_partition`, `EventProcessor::run`, and `PartitionClient::stream_events` now states this. ([#5094](https://github.com/Azure/azure-sdk-for-rust/issues/5094)) ### Other Changes diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs index 1e2252f2e84..55133aa846b 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/consumer/mod.rs @@ -228,9 +228,15 @@ impl ConsumerClient { }) } - /// Attaches a message receiver to a specific partition of the Event Hub. + /// Creates a message receiver for a specific partition of the Event Hub. /// - /// This function establishes a connection to the specified partition of the Event Hubs instance and returns a MessageReceiver which can be used to receive messages from it. + /// This function opens no AMQP link and does no network I/O. It builds an + /// [`EventReceiver`] from `options` and returns it. The AMQP link attaches on + /// the first poll of [`EventReceiver::stream_events`]. + /// + /// An unknown consumer group and an unknown partition id are reported from that + /// first poll, not from this call. A caller must poll the stream before it can + /// trust the receiver. /// /// # Arguments /// @@ -239,7 +245,7 @@ impl ConsumerClient { /// /// # Returns /// - /// A MessageReceiver which can be used to receive messages from the partition. + /// An [`EventReceiver`] which can be used to receive messages from the partition. /// /// Note that by default, a message receiver will receive events starting from the latest event in the partition (in /// other words, it will receive new events only). To receive events from another location within the partition you can @@ -344,7 +350,7 @@ impl ConsumerClient { consumer_group = %self.consumer_group, eventhub = %self.eventhub, source_url = %source_url, - "Receiver attached on partition." + "Created receiver on partition. The AMQP link attaches on the first stream_events() poll." ); Ok(EventReceiver::new( self.recoverable_connection.clone(), diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/partition_client.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/partition_client.rs index 83ff5ff82f0..33c79379a16 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/partition_client.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/partition_client.rs @@ -81,6 +81,10 @@ impl PartitionClient { /// This method returns a stream of `ReceivedEventData` wrapped in a `Result`. /// The stream yields events as they are received from the partition. /// + /// The partition's AMQP link attaches on the first poll of this stream. An unknown + /// consumer group and an unknown partition id are reported here, not from + /// [`EventProcessor::run`](crate::EventProcessor::run). + /// /// # Returns /// A stream of `Result` representing the received events. pub fn stream_events(&self) -> impl Stream> + '_ { diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/processor.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/processor.rs index 97102d90971..a5c9620b208 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/processor.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/processor.rs @@ -223,6 +223,11 @@ impl EventProcessor { /// to manage the ownership of partitions and distribute the load /// among consumers. /// The event processor will run until it is stopped or interrupted. + /// + /// Each partition receiver attaches its AMQP link on the first poll of + /// [`PartitionClient::stream_events`](crate::processor::PartitionClient::stream_events). + /// This method does not report an invalid consumer group. The first poll of the + /// partition stream reports it. /// # Errors /// Returns an error if the event processor fails to start. /// # Examples @@ -430,13 +435,14 @@ impl EventProcessor { )); } - // Since we can only have a single EventReceiver on a partition, we don't actually attempt to create the receiver until let start_position = self.get_start_position(&partition_id, checkpoints); debug!( partition_id = %partition_id, start_position = ?start_position, "Start position for partition." ); + // The AMQP link for this partition attaches on the first poll of + // `stream_events()`, so this call only builds the receiver. let receiver = self .consumer_client .open_receiver_on_partition(