fix(eventhubs): stop delivery and release ownership on shutdown

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md

@@ -18,6 +18,7 @@
 - `ConsumerClient::close` and `ProducerClient::close` now close the connection when another object still holds it, most often an `EventReceiver` that the caller has not dropped. Both methods used to report an error and leave the connection open. ([#4931](https://github.com/Azure/azure-sdk-for-rust/issues/4931))
 - A handle that outlives the client it came from now reports that the client is closed on its next call. Such a handle opened a second connection to the service before. ([#4931](https://github.com/Azure/azure-sdk-for-rust/issues/4931))
 - `EventProcessor::close` now continues past a partition client that the application still holds. It used to stop there, which left the partition clients behind it open and skipped the close of the consumer client. ([#4931](https://github.com/Azure/azure-sdk-for-rust/issues/4931))
+- `EventProcessor::shutdown` now stops the event delivery on every partition client that the processor issued, including a partition client that the application still holds, and it releases the ownership records of this instance, so another instance can claim those partitions without a wait for the expiration. It only set an internal flag before, so a held partition client kept delivering events after `run` returned. `close` runs the same stop path. ([#5096](https://github.com/Azure/azure-sdk-for-rust/issues/5096))
 - Claims-based-security authorizations for one connection now run in sequence. The service permits one `$cbs` link for each connection, so a client that attached more than one link at once could fail with `NotAllowed`.
 - `EventDataBatchOptions::max_size_in_bytes` now takes effect. A batch keeps the requested size, and `create_batch` reports an error when the request is zero or is larger than the sender link allows.
 - Increased `DEFAULT_PARTITION_EXPIRATION_DURATION` from 10 seconds to 60 seconds. The previous default was shorter than `DEFAULT_UPDATE_INTERVAL` (30 seconds), so ownership records expired between load-balancing cycles. The load balancer perpetually saw `current=0` for every consumer and continuously re-claimed partitions, causing widespread duplicate event processing. `EventProcessorBuilder::build` now rejects configurations where `partition_expiration_duration <= update_interval`. ([#3851](https://github.com/Azure/azure-sdk-for-rust/issues/3851))

sdk/eventhubs/azure_messaging_eventhubs/src/consumer/event_receiver.rs

@@ -282,6 +282,41 @@ impl Drop for EventReceiver {
     }
 }
 
+/// Builds an `EventReceiver` over a real `RecoverableConnection` whose
+/// next receiver attach fails with `attach_error`. No network activity
+/// happens: the injected error stops `ensure_receiver` before it opens
+/// a connection. It sits outside `mod tests` because the event processor
+/// tests need it too.
+#[cfg(test)]
+pub(crate) fn receiver_with_failing_attach(
+    partition_id: &str,
+    attach_error: AmqpError,
+) -> EventReceiver {
+    let source_url = Url::parse(&format!(
+        "amqps://example.servicebus.windows.net/eh/Partitions/{partition_id}"
+    ))
+    .unwrap();
+    let connection = RecoverableConnection::new(
+        Url::parse("amqps://example.servicebus.windows.net").unwrap(),
+        None,
+        None,
+        Arc::new(azure_core_test::credentials::MockCredential),
+        Default::default(),
+        None,
+    );
+    connection.force_attach_error(attach_error).unwrap();
+    EventReceiver::new(
+        connection,
+        AmqpReceiverOptions::default(),
+        AmqpSource::builder()
+            .with_address(source_url.to_string())
+            .build(),
+        source_url,
+        partition_id.to_string(),
+        None,
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -376,32 +411,6 @@ mod tests {
         );
     }
 
-    /// Builds an `EventReceiver` over a real `RecoverableConnection` whose
-    /// next receiver attach fails with `attach_error`. No network activity
-    /// happens: the injected error stops `ensure_receiver` before it opens
-    /// a connection.
-    fn receiver_with_failing_attach(attach_error: AmqpError) -> EventReceiver {
-        let connection = RecoverableConnection::new(
-            Url::parse("amqps://example.servicebus.windows.net").unwrap(),
-            None,
-            None,
-            Arc::new(azure_core_test::credentials::MockCredential),
-            Default::default(),
-            None,
-        );
-        connection.force_attach_error(attach_error).unwrap();
-        EventReceiver::new(
-            connection,
-            AmqpReceiverOptions::default(),
-            AmqpSource::builder()
-                .with_address(source_url().to_string())
-                .build(),
-            source_url(),
-            "0".to_string(),
-            None,
-        )
-    }
-
     // Drives the real stream. The function-level tests above prove what
     // `translate_attach_error` does when it is called; only this test proves
     // that `stream_events` calls it on the `get_receiver` failure path. If
@@ -411,7 +420,7 @@ mod tests {
     async fn stream_events_maps_stolen_attach_to_consumer_disconnected() {
         use futures::StreamExt;
 
-        let receiver = receiver_with_failing_attach(stolen());
+        let receiver = receiver_with_failing_attach("0", stolen());
         let mut stream = std::pin::pin!(receiver.stream_events());
         let error = stream
             .next()
@@ -431,7 +440,7 @@ mod tests {
     async fn stream_events_passes_other_attach_errors_through() {
         use futures::StreamExt;
 
-        let receiver = receiver_with_failing_attach(AmqpError::with_message("attach failed"));
+        let receiver = receiver_with_failing_attach("0", AmqpError::with_message("attach failed"));
         let mut stream = std::pin::pin!(receiver.stream_events());
         let error = stream
             .next()