Repository navigation
Python SDK: expose consumer shutdown and offset drain timeout #4165
Description
Description
The Rust high-level consumer exposes an asynchronous shutdown() method that drains pending offset work, performs final commits when enabled, leaves the consumer group, and stops its connection watcher.
The Python IggyConsumer currently has no corresponding method. Its consumer_group() builder also does not expose the Rust builder's offset_drain_timeout, which controls how long shutdown waits for background offset tasks.
This covers B22 and the currently applicable part of B23 in the Python SDK parity work tracked in #3893.
The commit_failed_messages setting listed in B23 is not present in the current Rust SDK and should not be added to the Python API.
Affected area / component
Python SDK
Proposed solution
- Add
IggyConsumer.shutdown()as an asynchronous method that locks the wrapped Rust consumer, calls itsshutdown(), and maps any error toRuntimeError. - Preserve the Rust method's idempotent behavior.
- Add an optional
offset_drain_timeout: datetime.timedelta | Noneargument toIggyClient.consumer_group(). - When provided, convert it to
IggyDurationand pass it toIggyConsumerBuilder::offset_drain_timeout(). - When omitted, preserve the Rust builder default of five seconds.
- Regenerate
foreign/python/apache_iggy.pyiand document the supported shutdown call order. - Extend
foreign/python/tests/test_consumer_group.pyto verify:- A consumer can be shut down after its consumption loop exits.
- Repeated shutdown calls succeed.
- A custom offset-drain timeout is accepted and propagated.
- Shutdown leaves a consumer group promptly so another member can receive its partitions.
Adding a replacement for the removed commit_failed_messages option is outside this issue.
Alternatives considered
Continue relying only on the asyncio.Event accepted by consume_messages(). That event stops one consumption loop but does not expose the Rust consumer's resource and offset cleanup operation.
Contribution
- I'm willing to submit a pull request to implement this feature
Good first issue
- I think this could be a good first issue for a new contributor
Activity
dedsec-terminal commentedon Sep 14, 2026
I would like to work on this. Opened #4184 with shutdown() and offset_drain_timeout for the Python consumer.
dedsec-terminal commentedon Sep 14, 2026
Could you assign this to me? I would like to pick this up.
justinmclean commentedon Sep 23, 2026
Difficulty: medium. The change is in the PyO3 binding (foreign/python/src/consumer.rs), and the producer's shutdown() in producer.rs is a working pattern to follow. The tests in foreign/python/tests/ run against a real server. Please read CONTRIBUTING.md first, in particular that PRs must be built and tested locally before they're opened. This is open for anyone to pick up.
shivaansh0610-LUFFY commentedon Sep 24, 2026
Hey @jiengup , I'd like to work on this too if it's still up for grabs - saw the note about
producer.rs'sshutdown()being the pattern to follow, makes sense as the template for locking the wrapped consumer and mapping errors toRuntimeError.Have gone through CONTRIBUTING.md, will run the full local check (
fmt/clippy/test/machete/sort/typos) and have a server up for theforeign/python/testssuite before opening anything.Happy to let you pick whoever you'd rather go with if someone's further along already, just wanted to throw my hat in.
slbotbm commentedon Sep 24, 2026
@shivaansh0610-LUFFY assigned
ParthibanRajasekaran commentedon Oct 26, 2026
@hubcio @justinmclean I noticed this issue is still open and that #4281 was closed after the earlier implementation became inactive.
My understanding is that the Python SDK needs to expose the Rust consumer’s existing shutdown lifecycle, including draining pending offsets and leaving the consumer group, plus an optional
offset_drain_timeoutonconsumer_group().In the current Python wrapper,
consume_messages()holds the consumer mutex for the duration of the run, while the Rust IggyConsumer::shutdown() is intended to be called after consumption stops. Would the clean approach be to keep the existing asyncio.Event as the way to stopconsume_messages(), then exposeshutdown()for the final drain and group leave?That would avoid the deadlock concern from #4281 while staying consistent with the Rust consumer API. I would also add tests proving that the group membership is released and that the final offset is stored. Would you prefer that lifecycle, or would you rather see a separate cancellation path that allows
shutdown()to interrupt an activeconsume_messages()task?If that direction sounds right, I’d be happy to take this issue and prepare a focused implementation :)
hubcio commentedon Oct 26, 2026
you can give it a shot. @slbotbm do u have any comment?
slbotbm commentedon Oct 26, 2026
@ParthibanRajasekaran can go ahead with the implementation