fix(eventhubs): add owner_level support, fix partition expiration, and add partition revocation

- Pass owner_level to AMQP receiver link as com.microsoft:epoch for
  broker-enforced exclusive partition access
- Fix default partition expiration from 0s to 60s so load balancer
  properly detects stale ownerships
- Add partition revocation mechanism: when dispatch() detects partitions
  lost to other consumers, it revokes the corresponding PartitionClient
  via an AtomicBool flag that callers check with is_revoked()
- Update processor example to demonstrate is_revoked() usage

Author: hjarraya

sdk/eventhubs/azure_messaging_eventhubs/examples/eventhubs_processor_client.rs

@@ -71,6 +71,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let checkpoint_store = Arc::new(InMemoryCheckpointStore::new());
     let processor = EventProcessor::builder()
+        // Setting owner_level enables epoch-based partition ownership.
+        // The Event Hub broker will disconnect any existing receiver on
+        // the same partition when a new receiver with the same or higher
+        // owner level connects, preventing duplicate event processing
+        // across multiple consumer instances.
+        .with_owner_level(0)
         .build(consumer, checkpoint_store)
         .await?;
     let background_processor = BackgroundProcessor::new(processor.clone());
@@ -90,6 +96,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut event_stream = partition_client.stream_events();
     let mut event_count = 0;
     while let Some(event) = event_stream.next().await {
+        // Check if the partition was reassigned to another consumer
+        // during rebalancing. When revoked, stop processing so the
+        // partition client can be re-acquired.
+        if partition_client.is_revoked() {
+            println!("Partition was reassigned, stopping.");
+            break;
+        }
         println!("Received message {event_count}");
         event_count += 1;
         if event_count > 10 {

sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/partition_client.rs

@@ -12,9 +12,12 @@ use azure_core_amqp::{message::AmqpAnnotationKey, AmqpValue};
 use futures::Stream;
 use std::{
     pin::Pin,
-    sync::{Arc, OnceLock, Weak},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, OnceLock, Weak,
+    },
 };
-use tracing::{debug, trace, warn};
+use tracing::{debug, info, trace, warn};
 
 /// Represents a client for interacting with a specific partition in Event Hubs.
 ///
@@ -26,6 +29,7 @@ pub struct PartitionClient {
     client_details: ConsumerClientDetails,
     event_receiver: OnceLock<EventReceiver>,
     consumers: Weak<ProcessorConsumersMap>,
+    revoked: Arc<AtomicBool>,
 }
 
 // It's safe to use the PartitionClient from multiple threads simultaneously.
@@ -45,6 +49,7 @@ impl PartitionClient {
             client_details,
             event_receiver: OnceLock::new(),
             consumers,
+            revoked: Arc::new(AtomicBool::new(false)),
         }
     }
 
@@ -56,6 +61,25 @@ impl PartitionClient {
         &self.partition_id
     }
 
+    /// Marks this partition client as revoked.
+    ///
+    /// When a partition is reassigned to another consumer by the load balancer,
+    /// the event processor calls this method to signal that this client should
+    /// stop processing events. Callers should check [`is_revoked`](Self::is_revoked)
+    /// between event receives and stop when it returns `true`.
+    pub fn revoke(&self) {
+        info!("Revoking partition client for partition {}", self.partition_id);
+        self.revoked.store(true, Ordering::Relaxed);
+    }
+
+    /// Returns `true` if this partition client has been revoked.
+    ///
+    /// A revoked partition client indicates that the partition has been
+    /// reassigned to another consumer and this client should stop processing.
+    pub fn is_revoked(&self) -> bool {
+        self.revoked.load(Ordering::Relaxed)
+    }
+
     /// Receives events from the partition.
     ///
     /// This method returns a stream of `ReceivedEventData` wrapped in a `Result`.

sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/processor.rs

@@ -50,6 +50,7 @@ pub struct EventProcessor {
     next_partition_client_sender: Sender<Arc<PartitionClient>>,
     client_details: ConsumerClientDetails,
     prefetch: u32,
+    owner_level: Option<i64>,
     update_interval: Duration,
     start_positions: StartPositions,
     is_running: std::sync::Mutex<bool>,
@@ -62,6 +63,7 @@ struct EventProcessorOptions {
     update_interval: Duration,
     start_positions: StartPositions,
     prefetch: u32,
+    owner_level: Option<i64>,
     partition_ids: Vec<String>,
 }
 
@@ -123,6 +125,38 @@ impl ProcessorConsumersMap {
         info!("Consumers for partition now: {:?}", consumers.keys());
         Ok(())
     }
+
+    /// Returns the set of partition IDs that have active partition clients.
+    fn get_active_partition_ids(&self) -> Result<Vec<String>> {
+        let consumers = self
+            .consumers
+            .lock()
+            .map_err(|_| EventHubsError::with_message("Could not lock consumers mutex."))?;
+        Ok(consumers.keys().cloned().collect())
+    }
+
+    /// Revokes and removes partition clients for partitions that are no longer
+    /// owned by this processor instance.
+    ///
+    /// This is called during dispatch when the load balancer indicates that
+    /// a partition has been reassigned to another consumer. The partition client
+    /// is marked as revoked (so the consumer can detect it via `is_revoked()`)
+    /// and removed from the consumers map (so a new client can be created if the
+    /// partition is later reassigned back).
+    fn revoke_partition_clients(&self, partition_ids: &[String]) -> Result<()> {
+        let mut consumers = self
+            .consumers
+            .lock()
+            .map_err(|_| EventHubsError::with_message("Could not lock consumers mutex."))?;
+        for partition_id in partition_ids {
+            if let Some(weak) = consumers.remove(partition_id) {
+                if let Some(client) = weak.upgrade() {
+                    client.revoke();
+                }
+            }
+        }
+        Ok(())
+    }
 }
 
 //pub(crate) type ConsumersType = std::sync::Mutex<HashMap<String, Arc<PartitionClient>>>;
@@ -163,6 +197,7 @@ impl EventProcessor {
             ))),
             client_details,
             prefetch: options.prefetch,
+            owner_level: options.owner_level,
             update_interval: options.update_interval,
             start_positions: options.start_positions,
             next_partition_client_sender: sender,
@@ -293,6 +328,25 @@ impl EventProcessor {
             e
         })?;
 
+        // Detect partitions that were stolen by another consumer.
+        // Compare the set of partitions we currently own (from load_balance)
+        // against active partition clients. Revoke any clients for partitions
+        // we no longer own so they stop processing.
+        let owned_ids: std::collections::HashSet<&str> =
+            ownerships.iter().map(|o| o.partition_id.as_str()).collect();
+        let active_ids = consumers.get_active_partition_ids()?;
+        let stolen: Vec<String> = active_ids
+            .into_iter()
+            .filter(|id| !owned_ids.contains(id.as_str()))
+            .collect();
+        if !stolen.is_empty() {
+            info!(
+                "Partitions no longer owned, revoking: {}",
+                stolen.join(", ")
+            );
+            consumers.revoke_partition_clients(&stolen)?;
+        }
+
         let checkpoints = self.get_checkpoint_map().await;
         let checkpoints = checkpoints.map_err(|e| {
             error!("Error in getting checkpoint map: {:?}", e);
@@ -366,6 +420,7 @@ impl EventProcessor {
                 Some(OpenReceiverOptions {
                     start_position: Some(start_position),
                     prefetch: Some(self.prefetch),
+                    owner_level: self.owner_level,
                     ..Default::default()
                 }),
             )
@@ -527,7 +582,7 @@ pub mod builders {
 
     const DEFAULT_PREFETCH: u32 = 300;
     const DEFAULT_UPDATE_INTERVAL: Duration = Duration::seconds(30);
-    const DEFAULT_PARTITION_EXPIRATION_DURATION: Duration = Duration::seconds(10);
+    const DEFAULT_PARTITION_EXPIRATION_DURATION: Duration = Duration::seconds(60);
 
     /// Builder for creating an `EventProcessor`.
     /// This builder allows you to configure various options for the event processor,
@@ -564,6 +619,7 @@ pub mod builders {
         start_positions: Option<StartPositions>,
         max_partition_count: Option<usize>,
         prefetch: Option<u32>,
+        owner_level: Option<i64>,
         load_balancing_strategy: Option<super::ProcessorStrategy>,
         partition_expiration_duration: Option<Duration>,
     }
@@ -611,6 +667,17 @@ pub mod builders {
             self
         }
 
+        /// Sets the owner level (epoch) for partition receivers.
+        ///
+        /// When set, the Event Hub broker enforces exclusive access: a new
+        /// receiver with the same or higher owner level will disconnect any
+        /// existing receiver on the same partition. This prevents duplicate
+        /// processing when partitions are rebalanced across consumers.
+        pub fn with_owner_level(mut self, owner_level: i64) -> Self {
+            self.owner_level = Some(owner_level);
+            self
+        }
+
         /// Sets the partition expiration duration for the event processor.
         pub fn with_partition_expiration_duration(
             mut self,
@@ -647,6 +714,7 @@ pub mod builders {
                     update_interval: self.update_interval.unwrap_or(DEFAULT_UPDATE_INTERVAL),
                     start_positions: self.start_positions.unwrap_or_default(),
                     prefetch: self.prefetch.unwrap_or(DEFAULT_PREFETCH),
+                    owner_level: self.owner_level,
                     partition_ids: eh_properties.partition_ids,
                 },
             )