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 83ff5ff82f0..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) @@ -230,3 +214,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}" + ); + } +}