From 839b849648a8a4e3d5922b429bf74f13a95cba7d Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 14:25:49 -0400 Subject: [PATCH 1/2] test(eventhubs): pin update_checkpoint annotation requirements The partition client returns Ok when an event carries no message annotations, and it writes a checkpoint with no offset and no sequence number when the annotations hold neither key. Both cases lose the caller's progress without a signal. Three tests fail against the unchanged source. A fourth does not compile, because it matches on an ErrorKind variant that the fix adds. Four more tests pass green. They pin the write path, the identity fields, and the store failure context. --- .../src/event_processor/partition_client.rs | 271 ++++++++++++++++++ 1 file changed, 271 insertions(+) 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..b9c45ee8565 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 @@ -230,3 +230,274 @@ impl Drop for PartitionClient { ); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::ErrorKind; + use crate::event_processor::Ownership; + use crate::in_memory_checkpoint_store::InMemoryCheckpointStore; + // Every AMQP name this module needs is declared here, not inherited from + // the parent module, so the tests survive a change to the parent imports. + use azure_core_amqp::{message::AmqpAnnotations, AmqpMessage, AmqpSymbol, AmqpValue}; + + const TEST_NAMESPACE: &str = "ns.servicebus.windows.net"; + const TEST_EVENT_HUB: &str = "test-eventhub"; + const TEST_CONSUMER_GROUP: &str = "test-consumer-group"; + + fn client_details() -> ConsumerClientDetails { + ConsumerClientDetails { + fully_qualified_namespace: TEST_NAMESPACE.to_string(), + consumer_group: TEST_CONSUMER_GROUP.to_string(), + eventhub_name: TEST_EVENT_HUB.to_string(), + client_id: "test-client".to_string(), + } + } + + fn client_with_store(partition_id: &str) -> (PartitionClient, Arc) { + let store = Arc::new(InMemoryCheckpointStore::new()); + let client = PartitionClient::new( + partition_id.to_string(), + store.clone(), + client_details(), + Weak::new(), + ); + (client, store) + } + + /// Builds an event whose AMQP message carries message annotations. An + /// empty slice still sets an empty annotation map, which is not the same + /// input as an absent map. + fn event_with(pairs: &[(&str, AmqpValue)]) -> ReceivedEventData { + let mut annotations = AmqpAnnotations::new(); + for (key, value) in pairs { + annotations.insert(AmqpSymbol::from(*key), value.clone()); + } + AmqpMessage::builder() + .with_message_annotations(annotations) + .build() + .into() + } + + fn event_without_annotations() -> ReceivedEventData { + AmqpMessage::default().into() + } + + struct FailingCheckpointStore; + + #[async_trait::async_trait] + impl CheckpointStore for FailingCheckpointStore { + async fn claim_ownership( + &self, + _ownerships: &[Ownership], + ) -> azure_core::Result> { + unreachable!("update_checkpoint must not claim ownership") + } + + async fn list_checkpoints( + &self, + _namespace: &str, + _event_hub_name: &str, + _consumer_group: &str, + ) -> azure_core::Result> { + unreachable!("update_checkpoint must not list checkpoints") + } + + async fn list_ownerships( + &self, + _namespace: &str, + _event_hub_name: &str, + _consumer_group: &str, + ) -> azure_core::Result> { + unreachable!("update_checkpoint must not list ownerships") + } + + async fn update_checkpoint(&self, _checkpoint: Checkpoint) -> azure_core::Result<()> { + Err(azure_core::Error::with_message( + azure_core::error::ErrorKind::Other, + "store is down", + )) + } + } + + #[tokio::test] + async fn update_checkpoint_rejects_an_event_without_message_annotations() { + let (client, store) = client_with_store("0"); + let event = event_without_annotations(); + + let result = client.update_checkpoint(&event).await; + let stored = store + .list_checkpoints(TEST_NAMESPACE, TEST_EVENT_HUB, TEST_CONSUMER_GROUP) + .await + .expect("the store must list its checkpoints"); + + assert!( + stored.is_empty(), + "an event without message annotations must write no checkpoint, got: {stored:?}" + ); + assert!( + result.is_err(), + "an event without message annotations must return an error to the caller" + ); + } + + #[tokio::test] + async fn update_checkpoint_rejects_an_event_without_offset_or_sequence_number() { + let (client, store) = client_with_store("1"); + let event = event_with(&[("x-opt-partition-key", AmqpValue::String("pk".into()))]); + + let result = client.update_checkpoint(&event).await; + let stored = store + .list_checkpoints(TEST_NAMESPACE, TEST_EVENT_HUB, TEST_CONSUMER_GROUP) + .await + .expect("the store must list its checkpoints"); + + assert!( + stored.is_empty(), + "annotations without an offset and without a sequence number must write no \ + checkpoint, got: {stored:?}" + ); + assert!( + result.is_err(), + "annotations without an offset and without a sequence number must return an error" + ); + } + + #[tokio::test] + async fn update_checkpoint_rejects_annotations_with_the_wrong_value_types() { + let (client, store) = client_with_store("2"); + let event = event_with(&[ + ("x-opt-offset", AmqpValue::Long(42)), + ("x-opt-sequence-number", AmqpValue::String("17".to_string())), + ]); + + let result = client.update_checkpoint(&event).await; + let stored = store + .list_checkpoints(TEST_NAMESPACE, TEST_EVENT_HUB, TEST_CONSUMER_GROUP) + .await + .expect("the store must list its checkpoints"); + + assert!( + stored.is_empty(), + "annotations with the wrong value types must write no checkpoint, got: {stored:?}" + ); + assert!( + result.is_err(), + "annotations with the wrong value types must return an error" + ); + } + + #[tokio::test] + async fn update_checkpoint_error_names_the_partition_and_the_kind() { + let (client, _store) = client_with_store("7"); + let event = event_with(&[]); + + let error = client + .update_checkpoint(&event) + .await + .expect_err("an event without the two annotations must return an error"); + + let ErrorKind::MissingCheckpointMetadata { partition_id } = &error.kind else { + panic!("the caller must be able to match on the kind, got: {error:?}"); + }; + assert_eq!( + partition_id.as_str(), + "7", + "the error must name the partition, got: {partition_id}" + ); + } + + #[tokio::test] + async fn update_checkpoint_writes_an_offset_only_checkpoint() { + let (client, store) = client_with_store("3"); + let event = event_with(&[("x-opt-offset", AmqpValue::String("1024".to_string()))]); + + client + .update_checkpoint(&event) + .await + .expect("an offset alone must write a checkpoint"); + + let stored = store + .list_checkpoints(TEST_NAMESPACE, TEST_EVENT_HUB, TEST_CONSUMER_GROUP) + .await + .expect("the store must list its checkpoints"); + assert_eq!(stored.len(), 1, "the store must hold one checkpoint"); + assert_eq!(stored[0].offset, Some("1024".to_string())); + assert_eq!(stored[0].sequence_number, None); + } + + #[tokio::test] + async fn update_checkpoint_writes_a_sequence_number_only_checkpoint() { + let (client, store) = client_with_store("4"); + let event = event_with(&[("x-opt-sequence-number", AmqpValue::Long(17))]); + + client + .update_checkpoint(&event) + .await + .expect("a sequence number alone must write a checkpoint"); + + let stored = store + .list_checkpoints(TEST_NAMESPACE, TEST_EVENT_HUB, TEST_CONSUMER_GROUP) + .await + .expect("the store must list its checkpoints"); + assert_eq!(stored.len(), 1, "the store must hold one checkpoint"); + assert_eq!(stored[0].offset, None); + assert_eq!(stored[0].sequence_number, Some(17)); + } + + #[tokio::test] + async fn update_checkpoint_writes_both_values_and_the_identity_fields() { + let (client, store) = client_with_store("5"); + let event = event_with(&[ + ("x-opt-offset", AmqpValue::String("2048".to_string())), + ("x-opt-sequence-number", AmqpValue::Long(99)), + ]); + + client + .update_checkpoint(&event) + .await + .expect("a complete pair of annotations must write a checkpoint"); + + let stored = store + .list_checkpoints(TEST_NAMESPACE, TEST_EVENT_HUB, TEST_CONSUMER_GROUP) + .await + .expect("the store must list its checkpoints"); + assert_eq!(stored.len(), 1, "the store must hold one checkpoint"); + assert_eq!(stored[0].fully_qualified_namespace, TEST_NAMESPACE); + assert_eq!(stored[0].event_hub_name, TEST_EVENT_HUB); + assert_eq!(stored[0].consumer_group, TEST_CONSUMER_GROUP); + assert_eq!(stored[0].partition_id, "5"); + assert_eq!(stored[0].offset, Some("2048".to_string())); + assert_eq!(stored[0].sequence_number, Some(99)); + } + + #[tokio::test] + async fn update_checkpoint_reports_a_store_failure_with_its_context() { + let client = PartitionClient::new( + "6".to_string(), + Arc::new(FailingCheckpointStore), + client_details(), + Weak::new(), + ); + let event = event_with(&[ + ("x-opt-offset", AmqpValue::String("4096".to_string())), + ("x-opt-sequence-number", AmqpValue::Long(5)), + ]); + + let error = client + .update_checkpoint(&event) + .await + .expect_err("a store failure must reach the caller"); + + assert!( + matches!(error.kind, ErrorKind::AzureCore(_)), + "a store failure must keep the Azure Core kind, got: {error:?}" + ); + assert!( + error + .to_string() + .contains("Failed to update checkpoint for partition 6"), + "the error must name the partition, got: {error}" + ); + } +} From e1f9b12af1d56f2436c6ff556be7ae5710ba9223 Mon Sep 17 00:00:00 2001 From: Johnathan W Date: Thu, 20 Aug 2026 14:36:10 -0400 Subject: [PATCH 2/2] fix(eventhubs): reject a checkpoint from an event with no position PartitionClient::update_checkpoint returned Ok(()) and wrote nothing when the event had no message annotations. It also wrote a checkpoint with no offset and no sequence number when the annotations held neither value. Such a checkpoint names no position in the partition. It suppressed the per-partition start position the caller configured, because EventProcessor prefers any stored checkpoint over that position. It also erased a good checkpoint in BlobCheckpointStore, because the store builds the blob metadata from the checkpoint fields, and Azure Blob Storage replaces all metadata on a set-metadata call. The method now reads the offset and the sequence number through the ReceivedEventData accessors and returns the new error variant ErrorKind::MissingCheckpointMetadata when both are absent. An event that carries only one of the two still writes a checkpoint. This matches the InvalidOperationException that .NET raises for the same input. Refs #5097 --- .../azure_messaging_eventhubs/CHANGELOG.md | 3 ++ .../azure_messaging_eventhubs/src/error.rs | 21 ++++++++ .../src/event_processor/partition_client.rs | 50 +++++++------------ 3 files changed, 41 insertions(+), 33 deletions(-) diff --git a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md index 8575377c254..4241a940562 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md +++ b/sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md @@ -8,10 +8,12 @@ - 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)` error variant. - Added the `ErrorKind::InvalidBatchSize { requested, max_allowed }` error variant. `create_batch` reports it when `EventDataBatchOptions::max_size_in_bytes` is zero or is larger than the maximum the sender link allows, so a caller can branch on the kind instead of the message. This matches the `ArgumentOutOfRangeException` that .NET raises and the typed error that Go returns for the same input. +- Added the `ErrorKind::MissingCheckpointMetadata { partition_id }` error variant. `PartitionClient::update_checkpoint` reports it when the event carries no offset and no sequence number, so a caller can branch on the kind instead of the message. This matches the `InvalidOperationException` that .NET raises for the same input. ### Breaking Changes - On the receive path, the `amqp:link:stolen` AMQP condition is no longer auto-retried. A receiver displaced by a higher-or-equal-epoch attacher now surfaces the error (translated to `EventHubsError::ConsumerDisconnected` by `EventReceiver::stream_events`) instead of silently re-attaching. Sender, CBS, and management operations retain the historical retry-on-stolen behavior. +- `PartitionClient::update_checkpoint` now returns an error when the event carries no offset and no sequence number. Such a call returned `Ok(())` and recorded nothing before. ### Bugs Fixed @@ -26,6 +28,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)) +- `PartitionClient::update_checkpoint` no longer reports success without writing a checkpoint. It returned `Ok(())` and wrote nothing when the event had no message annotations. It also wrote a checkpoint with no offset and no sequence number when the annotations held neither value. Such a checkpoint suppressed the per-partition start position the caller configured, because `EventProcessor` prefers any stored checkpoint over that position. It also erased a good checkpoint in `BlobCheckpointStore`, because the store builds the blob metadata from the checkpoint fields, and Azure Blob Storage replaces all metadata on a set-metadata call. ([#5097](https://github.com/Azure/azure-sdk-for-rust/issues/5097)) ### Other Changes diff --git a/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs b/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs index 38c7f126592..ab653dc15f2 100644 --- a/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs +++ b/sdk/eventhubs/azure_messaging_eventhubs/src/error.rs @@ -51,6 +51,20 @@ pub enum ErrorKind { /// `matches!(err.kind, ErrorKind::ConsumerDisconnected(_))`. /// Mirrors `EventHubsException.FailureReason.ConsumerDisconnected` (.NET). ConsumerDisconnected(Option), + + /// The event carries no offset and no sequence number, so it names no + /// position in the partition. A checkpoint built from such an event holds + /// no position, and it erases the position the checkpoint store already + /// holds. + /// + /// Mirrors the `InvalidOperationException` that .NET raises for the same + /// input ("A checkpoint cannot be created or updated using an empty + /// event."). Match on the variant to tell it apart from a store failure: + /// `matches!(err.kind, ErrorKind::MissingCheckpointMetadata { .. })`. + MissingCheckpointMetadata { + /// The identifier of the partition the checkpoint is for. + partition_id: String, + }, } /// Represents an error that can occur in the Event Hubs module. @@ -102,6 +116,13 @@ impl std::fmt::Display for EventHubsError { e ) } + ErrorKind::MissingCheckpointMetadata { partition_id } => write!( + f, + "Cannot record a checkpoint for partition {}. \ + The event carries no offset and no sequence number, \ + so there is nothing to record.", + partition_id + ), } } } 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 b9c45ee8565..754376cf71e 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 @@ -3,12 +3,11 @@ use super::processor::ProcessorConsumersMap; use crate::{ - error::Result, + error::{ErrorKind, Result}, models::{Checkpoint, ConsumerClientDetails, ReceivedEventData}, processor::CheckpointStore, EventHubsError, EventReceiver, }; -use azure_core_amqp::{message::AmqpAnnotationKey, AmqpValue}; use futures::Stream; use std::{ pin::Pin, @@ -144,45 +143,30 @@ impl PartitionClient { /// Updates the checkpoint for the current partition. /// - /// This method extracts the sequence number and offset from the provided `ReceivedEventData` + /// This method reads the offset and the sequence number from the provided `ReceivedEventData` /// and updates the checkpoint in the `CheckpointStore`. /// /// # Arguments - /// * `event_data` - The event data containing the sequence number and offset to update the checkpoint. + /// * `event_data` - The event data that carries the offset and the sequence number to record. /// /// # Errors - /// Returns an error if the sequence number or offset is invalid, or if updating the checkpoint fails. + /// Returns [`ErrorKind::MissingCheckpointMetadata`](crate::error::ErrorKind::MissingCheckpointMetadata) + /// when the event carries no offset and no sequence number. Such an event names no position in + /// the partition, and a checkpoint with both fields empty erases the position the store holds. + /// Returns an error also when the checkpoint store fails to write the checkpoint. pub async fn update_checkpoint(&self, event_data: &ReceivedEventData) -> Result<()> { - let mut offset_option = None; - let mut sequence_number_option = None; - - let event_data_message = event_data.raw_amqp_message(); - let Some(message_annotations) = event_data_message.message_annotations.as_ref() else { - // No message annotations. Nothing to do. - return Ok(()); - }; - for (key, value) in message_annotations.0.iter() { - let AmqpAnnotationKey::Symbol(symbol) = key else { - continue; - }; - - if *symbol == "x-opt-offset" { - let AmqpValue::String(offset_value) = value else { - continue; - }; - offset_option = Some(offset_value.clone()); - } else if *symbol == "x-opt-sequence-number" { - let AmqpValue::Long(sequence_number_value) = value else { - continue; - }; - sequence_number_option = Some(*sequence_number_value); - } + let offset = event_data.offset().clone(); + let sequence_number = event_data.sequence_number(); + if offset.is_none() && sequence_number.is_none() { + return Err(EventHubsError::from(ErrorKind::MissingCheckpointMetadata { + partition_id: self.partition_id.clone(), + })); } debug!( partition_id = %self.partition_id, - sequence_number = ?sequence_number_option, - offset = ?offset_option, + sequence_number = ?sequence_number, + offset = ?offset, "Updating checkpoint for partition." ); let checkpoint = Checkpoint { @@ -190,8 +174,8 @@ impl PartitionClient { event_hub_name: self.client_details.eventhub_name.clone(), consumer_group: self.client_details.consumer_group.clone(), partition_id: self.partition_id.clone(), - offset: offset_option, - sequence_number: sequence_number_option, + offset, + sequence_number, }; self.checkpoint_store .update_checkpoint(checkpoint)