fix(eventhubs): reject a checkpoint from an event with no position

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

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<AmqpDescribedError>),
+
+    /// 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
+            ),
         }
     }
 }