From 6a63444a81cf2add3232135735d72f4c64a56c47 Mon Sep 17 00:00:00 2001 From: ParthibanRajasekaran Date: Tue, 27 Oct 2026 11:04:15 +0000 Subject: [PATCH 1/4] feat(python): expose consumer shutdown and offset drain timeout --- foreign/python/apache_iggy.pyi | 16 ++++++++++++++++ foreign/python/src/client.rs | 11 +++++++++++ foreign/python/src/consumer.rs | 20 ++++++++++++++++++++ 3 files changed, 47 insertions(+) diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index cfaef53419..f119f7635b 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -2103,6 +2103,7 @@ class IggyClient: init_retries: builtins.int | None = None, init_retry_interval: datetime.timedelta | None = None, allow_replay: builtins.bool = False, + offset_drain_timeout: datetime.timedelta | None = None, ) -> collections.abc.Awaitable[IggyConsumer]: r""" Creates a new consumer group consumer. @@ -2113,6 +2114,11 @@ class IggyClient: `AutoCommit` interval is negative, or if any of those except `poll_interval` is zero. + `offset_drain_timeout=None` preserves the Rust default of five seconds for each + background offset task during shutdown. Zero aborts tasks that are still pending; + final commits and group leave still run. Negative values raise `ValueError`. + This timeout does not bound the entire shutdown operation. + Consumer groups are not available over HTTP. With `auto_join_consumer_group` left on, this call fails at the join with `Feature is unavailable`. Turning it off is not a workaround: the join is skipped, but a group @@ -2260,6 +2266,15 @@ class IggyConsumer: consumer group member. The consumer drops the failed route and the position it continued from, and the next iteration routes again. """ + def shutdown(self) -> collections.abc.Awaitable[None]: + r""" + Drains pending offset work, performs final commits when auto-commit is enabled, + leaves the consumer group and stops its connection watcher. Repeated calls are safe. + Stop `consume_messages()` with its `asyncio.Event` and await its completion first. + Do not await this method from the callback or while an iterator's next call is pending: + consumption holds the same consumer lock. Subsequent iterator calls terminate. + The Rust SDK currently logs final commit and group-leave failures rather than raising them. + """ def consume_messages( self, callback: collections.abc.Callable[ @@ -2270,6 +2285,7 @@ class IggyConsumer: r""" Consumes messages continuously using a callback function and an optional `asyncio.Event` for signaling shutdown. Returns an awaitable that completes when shutdown is signaled or a RuntimeError on failure. + After it completes, call `shutdown()` to drain offsets and release group membership. After another client deleted and created a partition again, one poll can fail with `HistoryUnavailable` (87), or `ConsumerGroupPartitionNotOwned` (5009) for a consumer group member. The error is logged, not raised: the failed route is dropped, and the diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 34a0dfb5ac..9d80fc4245 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -1778,6 +1778,11 @@ impl IggyClient { /// `AutoCommit` interval is negative, or if any of those except `poll_interval` /// is zero. /// + /// `offset_drain_timeout=None` preserves the Rust default of five seconds for each + /// background offset task during shutdown. Zero aborts tasks that are still pending; + /// final commits and group leave still run. Negative values raise `ValueError`. + /// This timeout does not bound the entire shutdown operation. + /// /// Consumer groups are not available over HTTP. With `auto_join_consumer_group` /// left on, this call fails at the join with `Feature is unavailable`. /// Turning it off is not a workaround: the join is skipped, but a group @@ -1802,6 +1807,7 @@ impl IggyClient { init_retries=None, init_retry_interval=None, allow_replay=false, + offset_drain_timeout=None, ))] #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[IggyConsumer]", imports=("collections.abc")))] fn consumer_group<'a>( @@ -1828,6 +1834,8 @@ impl IggyClient { #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] init_retry_interval: Option>, allow_replay: bool, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + offset_drain_timeout: Option>, ) -> PyResult> { let mut builder = self .inner @@ -1889,6 +1897,9 @@ impl IggyClient { if allow_replay { builder = builder.allow_replay() } + if let Some(timeout) = offset_drain_timeout { + builder = builder.offset_drain_timeout(py_delta_to_iggy_duration(&timeout)?); + } let mut consumer = builder.build(); future_into_py(py, async move { diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs index 0dd77e0ce6..937f2e0127 100644 --- a/foreign/python/src/consumer.rs +++ b/foreign/python/src/consumer.rs @@ -189,8 +189,28 @@ impl IggyConsumer { ReceiveMessageIterator { inner } } + /// Drains pending offset work, performs final commits when auto-commit is enabled, + /// leaves the consumer group and stops its connection watcher. Repeated calls are safe. + /// Stop `consume_messages()` with its `asyncio.Event` and await its completion first. + /// Do not await this method from the callback or while an iterator's next call is pending: + /// consumption holds the same consumer lock. Subsequent iterator calls terminate. + /// The Rust SDK currently logs final commit and group-leave failures rather than raising them. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn shutdown<'a>(&self, py: Python<'a>) -> PyResult> { + let inner = self.inner.clone(); + future_into_py(py, async move { + inner + .lock() + .await + .shutdown() + .await + .map_err(to_runtime_error) + }) + } + /// Consumes messages continuously using a callback function and an optional `asyncio.Event` for signaling shutdown. /// Returns an awaitable that completes when shutdown is signaled or a RuntimeError on failure. + /// After it completes, call `shutdown()` to drain offsets and release group membership. /// After another client deleted and created a partition again, one poll can fail with /// `HistoryUnavailable` (87), or `ConsumerGroupPartitionNotOwned` (5009) for a consumer /// group member. The error is logged, not raised: the failed route is dropped, and the From 5a0256353ced257fa330da58dbcb2a358780d542 Mon Sep 17 00:00:00 2001 From: ParthibanRajasekaran Date: Tue, 27 Oct 2026 11:04:24 +0000 Subject: [PATCH 2/4] test(python): verify consumer shutdown lifecycle --- foreign/python/tests/test_consumer_group.py | 126 ++++++++++++++++++++ 1 file changed, 126 insertions(+) diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py index f8c44720f0..6ae9486617 100644 --- a/foreign/python/tests/test_consumer_group.py +++ b/foreign/python/tests/test_consumer_group.py @@ -2164,3 +2164,129 @@ async def take(_: ReceiveMessage) -> None: shutdown_event.set() await consumer.consume_messages(take, shutdown_event) + + +class TestConsumerShutdown: + @pytest.mark.asyncio + @pytest.mark.parametrize( + "timeout_options", + [ + {}, + {"offset_drain_timeout": None}, + {"offset_drain_timeout": timedelta(seconds=2)}, + {"offset_drain_timeout": timedelta(0)}, + ], + ids=["omitted", "none", "custom", "zero"], + ) + async def test_shutdown_drains_offsets_and_releases_partition( + self, iggy_client: IggyClient, unique_name, timeout_options + ): + stream, topic, group = unique_name(), unique_name(), unique_name() + first = replacement = replacement_client = consume_task = None + stop = asyncio.Event() + try: + async with asyncio.timeout(15): + await iggy_client.create_stream(stream) + await iggy_client.create_topic(stream, topic) + await iggy_client.send_messages( + stream, topic, 0, [Message(str(i)) for i in range(3)] + ) + first = await iggy_client.consumer_group( + group, + stream, + topic, + polling_strategy=PollingStrategy.First(), + batch_length=1, + auto_commit=AutoCommit.Interval(timedelta(hours=1)), + poll_interval=timedelta(milliseconds=10), + **timeout_options, + ) + await wait_for_consumer_group_assignment( + iggy_client, stream, topic, group, members_count=1 + ) + received = [] + + async def take(message: ReceiveMessage) -> None: + received.append(message) + if len(received) == 3: + stop.set() + + consume_task = asyncio.ensure_future(first.consume_messages(take, stop)) + await asyncio.shield(consume_task) + assert [message.payload().decode() for message in received] == [ + "0", + "1", + "2", + ] + partition = received[-1].partition_id() + final_offset = received[-1].offset() + assert final_offset > 0 + assert first.get_last_consumed_offset(partition) == final_offset + stored = first.get_last_stored_offset(partition) + assert stored is None or stored < final_offset + + await asyncio.gather(*(first.shutdown() for _ in range(3))) + assert first.get_last_stored_offset(partition) == final_offset + details = await iggy_client.get_consumer_group(stream, topic, group) + assert details is not None + assert details.members_count == 0 + assert details.members == [] + await first.shutdown() + with pytest.raises(StopAsyncIteration): + await first.iter_messages().__anext__() + + await iggy_client.send_messages(stream, topic, 0, [Message("3")]) + + # A separate connection cannot reuse the first member's membership. + host, port = get_server_config() + replacement_client = IggyClient(f"{host}:{port}") + await replacement_client.connect() + await replacement_client.login_user("iggy", "iggy") + replacement = await replacement_client.consumer_group( + group, + stream, + topic, + polling_strategy=PollingStrategy.Next(), + batch_length=1, + auto_commit=AutoCommit.Disabled(), + poll_interval=timedelta(milliseconds=10), + ) + await wait_for_consumer_group_assignment( + iggy_client, stream, topic, group, members_count=1 + ) + message = await replacement.iter_messages().__anext__() + assert message.partition_id() == partition + assert message.offset() == final_offset + 1 + assert message.payload().decode() == "3" + finally: + stop.set() + try: + if consume_task is not None: + await asyncio.wait_for(asyncio.shield(consume_task), timeout=5) + finally: + try: + await asyncio.wait_for( + asyncio.gather( + *( + consumer.shutdown() + for consumer in (replacement, first) + if consumer is not None + ) + ), + timeout=5, + ) + finally: + if replacement_client is not None: + await asyncio.wait_for(replacement_client.shutdown(), timeout=5) + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_negative_offset_drain_timeout(self): + client = IggyClient("127.0.0.1:8090") + with pytest.raises(ValueError, match="duration must not be negative"): + await client.consumer_group( + "group", + "stream", + "topic", + offset_drain_timeout=timedelta(microseconds=-1), + ) From 82e531057427f75751155e6ab9ca079d16d1bf7b Mon Sep 17 00:00:00 2001 From: ParthibanRajasekaran Date: Tue, 27 Oct 2026 11:04:25 +0000 Subject: [PATCH 3/4] docs(python): explain consumer shutdown order and timeout --- foreign/python/README.md | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/foreign/python/README.md b/foreign/python/README.md index a7ac8f6256..a32094094c 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -554,6 +554,42 @@ accepted messages before returning. Object destruction does not perform asynchronous cleanup; dropping a background producer without `shutdown()` can lose buffered messages. +## Consumer Shutdown + +Stop a callback consumption loop with its `asyncio.Event`, await the loop, then +shut down the consumer before shutting down its client: + +```python +stop = asyncio.Event() + + +async def take(message): + print(message.payload()) + stop.set() + + +await consumer.consume_messages(take, stop) +await consumer.shutdown() +await client.shutdown() +``` + +When another task requests the stop, call `stop.set()` there and await the +existing consumption task before `consumer.shutdown()`. Do not await shutdown +inside the callback or while an iterator's next call is pending: they hold the +same consumer lock. After shutdown, iterators terminate. Repeated and concurrent +shutdown calls are safe once consumption has stopped. + +Shutdown drains background offset work, performs final commits when auto-commit +is enabled, leaves the group and stops the connection watcher. The Rust SDK logs +final commit and group-leave failures; successful return does not confirm these +operations succeeded on the server. `AutoCommit.Disabled()` skips final commits. + +`consumer_group(..., offset_drain_timeout=timedelta(seconds=2))` overrides the +Rust default of five seconds per background offset task. Omit it or pass `None` +to retain that default. Zero aborts background work that remains pending, but +final commits and group leave still run. Negative values raise `ValueError`. +This is not a deadline for the entire shutdown operation. + ## Examples Refer to the [examples/python/](https://github.com/apache/iggy/tree/master/examples/python) directory for usage examples. From 62356f2ae675472b727c28db12ad4518d9576b03 Mon Sep 17 00:00:00 2001 From: ParthibanRajasekaran Date: Tue, 27 Oct 2026 11:35:15 +0000 Subject: [PATCH 4/4] test(python): adapt shutdown tests to deferred polling --- foreign/python/tests/test_consumer_group.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py index 3a29526e9b..80dd30973c 100644 --- a/foreign/python/tests/test_consumer_group.py +++ b/foreign/python/tests/test_consumer_group.py @@ -2162,7 +2162,6 @@ async def test_shutdown_drains_offsets_and_releases_partition( polling_strategy=PollingStrategy.First(), batch_length=1, auto_commit=AutoCommit.Interval(timedelta(hours=1)), - poll_interval=timedelta(milliseconds=10), **timeout_options, ) await wait_for_consumer_group_assignment( @@ -2213,7 +2212,6 @@ async def take(message: ReceiveMessage) -> None: polling_strategy=PollingStrategy.Next(), batch_length=1, auto_commit=AutoCommit.Disabled(), - poll_interval=timedelta(milliseconds=10), ) await wait_for_consumer_group_assignment( iggy_client, stream, topic, group, members_count=1