fix(python): read consumer metadata without the consumer lock - #3888
fix(python): read consumer metadata without the consumer lock#3888ethanlin01x wants to merge 13 commits into
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3888 +/- ##
============================================
- Coverage 84.03% 84.03% -0.01%
Complexity 1358 1358
============================================
Files 1215 1215
Lines 170820 170847 +27
Branches 138576 138603 +27
============================================
+ Hits 143547 143566 +19
+ Misses 23411 23398 -13
- Partials 3862 3883 +21
🚀 New features to boost your workflow:
|
8d0d07d to
b73cf67
Compare
|
/request-review @hubcio |
|
/ready |
7b7a74e to
5763540
Compare
The synchronous getters on IggyConsumer took the consumer mutex with blocking_lock() while holding the GIL, and consume_messages holds that mutex for the whole consumption run. Reading an attribute during consumption hung the interpreter; reading one from a callback panicked inside the Tokio runtime. None of those getters need exclusive access. The name, stream and topic are fixed at construction, and the partition id and offsets live behind Arcs that IggyConsumerState now exposes as a cloneable view. IggyConsumer owns that state and delegates to it, so the Python wrapper reads metadata without the lock. Found while reviewing apache#3776.
5763540 to
e9c641e
Compare
hubcio
left a comment
There was a problem hiding this comment.
fix is good, some polishing is needed tho.
three that don't fit on a diff line:
foreign/python/src/consumer.rs:410-411: thelock_owned()guard is shadowed, not dropped, so it spans the wholeconsume()including the callback await. harmless today, butself.task_locals.lock().await.clone()drops it.foreign/python/src/consumer.rs:202:tokio::join!waits on the shutdown task even when consume already exited, and that task waits forever onEvent.wait(). the new test is only safe because it setsshutdown_eventinfinally.- nothing regenerates and diffs
apache_iggy.pyiin ci - php has that gate, python doesn't. no impact here, separate ticket.
|
@hubcio do we want to check the pyi file in ci? i ask because this check will require a complete recompilation of the sdk, and our ci runs are very heavy as it is. maybe we could instead redesign the ci for ffi sdks instead? |
|
@slbotbm like we discussed on discord, please propose something via issues or discussions. |
The flush loops held a DashMap shard read guard across the store await while the poll future can insert a new partition into the same map.
A consumption run holds the consumer mutex while it awaits the callback, so awaiting store_offset or delete_offset from inside one deadlocked. Under AutoCommit.Disabled() that left no manual commit path at all. Both take &self in the Rust SDK, so IggyConsumerState now owns them and the Python wrapper reads through it instead of taking the lock.
tokio::join! waited on a task that parks on Event.wait() forever once consuming has returned on its own, so it is now aborted instead. Signalling a receiver that is already gone is no longer an error, or it would mask the result of the run it was trying to stop. The task locals guard was shadowed rather than dropped, so it spanned the whole consume including the callback await.
IggyConsumerState is public through the prelude and had no Debug. `None` from the offset getters means the partition is untracked, not that nothing was consumed, and stream() / topic() give back an identifier rather than a name.
faulthandler_timeout and faulthandler_exit_on_timeout give the whole suite what one test body was doing by hand.
# Conflicts: # core/sdk/src/clients/consumer.rs # foreign/python/src/consumer.rs
Rust 1.98's clippy flags the enum for a large size difference between variants under -D warnings.
|
@hubcio Thanks for the review! Also fixed a clippy error while at it. |
Which issue does this PR address?
Relates to #3776 (Found while reviewing)
Rationale
The synchronous getters on the Python
IggyConsumertook the consumer mutex withblocking_lock()while holding the GIL, andconsume_messagesholds that mutex for the whole consumption run. Reading an attribute during consumption hung the interpreter; reading one from a callback panicked inside the Tokio runtime. Neither is recoverable from Python.What changed?
None of those getters need exclusive access.
name,streamandtopicnever change after construction, so the Python wrapper snapshots them. The partition id and offsets do change, but already live behindArcs in the Rust SDK, which now hands them out asIggyConsumerStateviaIggyConsumer::state(). The wrapper keeps a clone and reads them as atomic loads, never taking the lock.Additive on the Rust side. On the Python side
stream()andtopic()now return the identifier directly instead of aPyResult.Local Execution
AI Usage