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

[Johnathan W (j7nw4r)]

Summary

EventProcessor::shutdown now stops the event delivery and releases the ownership records that the instance holds. It set an internal flag and did nothing else before, so a partition client that the application already held kept delivering events after run returned.

Motivation

shutdown set is_running = false and stopped there. run built its consumers map as a local value and stored it nowhere, so shutdown could reach no receiver at all. Every receiver stayed attached, and every ownership record stayed live, so a second instance could not claim those partitions until the records expired. close closed receivers, but it consumes the processor out of its Arc, which a caller that holds a partition client cannot do.

shutdown now guarantees two things. It stops the delivery on every partition client that the processor issued, including one the application still holds, and the stream_events stream of such a client resolves with ConsumerDisconnected. It then releases the ownership records whose owner is this instance, so another instance can claim those partitions without a wait for the expiration. A failed release logs at the warn level and does not fail the call.

close differs from shutdown in what it adds, not in what it stops. close runs the same stop path, and it also consumes the processor, drains the queued partition clients, and closes the consumer client. close is therefore a superset of shutdown.

Changes

• EventProcessor keeps its consumers map on self, and run uses that map in place of a local one.
• Added ProcessorConsumersMap::close_all_receivers, which closes each receiver and removes no map entry, so a partition client that the application retains keeps its place.
• Added a private stop that shutdown and close share. It drops the is_running guard before it awaits, because a std::sync::MutexGuard is not Send and holding it across an await deadlocks.
• The ownership release keeps only the records whose owner id matches this instance, and it keeps the ETag that the store returned, because claim_ownership rejects a stale ETag.
• Documented that run reads the shutdown flag only after the update_interval sleep, so run can take up to one full update_interval to return, while the delivery stops as soon as shutdown returns.
• Added a CHANGELOG entry under the unreleased version.

Test plan

• Five offline tests in the mod tests of processor.rs were written first and proved red against the unchanged source: shutdown_closes_receivers_of_issued_partition_clients, shutdown_releases_only_this_instances_ownerships, shutdown_continues_when_the_ownership_release_fails, shutdown_twice_succeeds_and_keeps_ownership_released, and close_runs_the_shutdown_stop_path.
• Each delivery test installs an offline EventReceiver before it asserts, because a partition client with an empty receiver returns a canned error stream that pins nothing. An unclosed offline receiver yields AmqpError, and only the closed path yields ConsumerDisconnected(None).
• shutdown_future_is_send guards the deadlock. It becomes a compile error, not a failed assertion, if a later change holds the is_running guard across an await.
• close_continues_past_a_retained_partition_client still passes, which holds the contract that a retained partition client stays in the consumers map.
• CARGO_BUILD_JOBS=1 cargo test --package azure_messaging_eventhubs --lib -- --test-threads=1 reports 150 passed, 0 failed, 14 ignored.
• CARGO_BUILD_JOBS=1 RUSTFLAGS=-Dwarnings cargo test --no-run --package azure_messaging_eventhubs exits 0. cargo fmt --check and cargo clippy --all-targets are clean.
• Assertions were added inside the existing live test receive_events_from_processor, which asserts that a retained partition client stops within 30 seconds of shutdown. No live test ran in this session, so that assertion waits for a live pass.

Closes #5096

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
3 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).
3 pipeline(s) were filtered out due to trigger conditions.
There may be pipelines that require an authorized user to comment /azp run to run.

[Copilot]

🟡 Changes recommended

Shutdown can race with an active dispatch, allowing delivery and ownership claims after the stop sequence.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Updates EventProcessor shutdown behavior to stop event delivery and release partition ownership promptly.

Changes:

• Shares the consumer map across processor lifecycle operations.
• Adds receiver shutdown and ownership release logic with tests.
• Documents the behavior and updates the changelog.

File summaries

File	Description
src/event_processor/processor.rs	Implements shared stop logic and tests.
src/consumer/event_receiver.rs	Exposes a test-only receiver helper.
tests/eventhubs_processor.rs	Verifies retained-client shutdown behavior.
CHANGELOG.md	Records the shutdown fix.

Review details

• Files reviewed: 4/4 changed files
• Comments generated: 1
• Review effort level: Balanced

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[Sagar Patel (sagar0207)]

One lifecycle edge case looks worth covering before merge: reusing the same EventProcessor after shutdown().

Before this change, every run() created a fresh ProcessorConsumersMap. The map now lives on EventProcessor, and close_all_receivers() deliberately retains its entries. add_partition_client() treats any existing key as a live client, even when the stored Weak<PartitionClient> no longer upgrades, while a retained client's EventReceiver has been permanently closed and cannot be replaced because it is stored in a OnceLock.

As a result, a subsequent run() can reclaim ownership but skip creating usable partition clients. Is shutdown() intended to be terminal for an EventProcessor instance? If so, could that be documented and enforced by rejecting another run()? If restart is intended to work, the map needs generation/identity-aware cleanup (so an old client cannot later remove a replacement), plus a shutdown -> run again test.