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.