diff --git a/foreign/python/README.md b/foreign/python/README.md index e8302f6cf7..b8c39fd867 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -650,11 +650,16 @@ when you call `mark(checkpoint)` with `CommitAfter.Marked`. `AutoCommit.default()` stores every 10,000 messages or 5 seconds after processing. With `CommitAfter.Processing`, a `consume_messages()` callback that raises ends the run with its exception, and its message is read again by the -next run. A reader takes the same `commit`. Call `shutdown()` before you drop the consumer, or use it in -`async with`, which shuts it down. `shutdown()` also ends an `async for` that -waits for a message. In a `consume_messages()` callback, `shutdown()`, iterating -the consumer and a nested `consume_messages()` raise `RuntimeError`, because the -run waits for the callback: set the shutdown event instead. +next run. A reader takes the same `commit`. Call `shutdown()` before you drop the +consumer, or use it in `async with`, which shuts it down. Shutdown waits up to +five seconds for pending automatic offset stores by default. Pass +`offset_drain_timeout` to `consumer_group()` to change that wait; zero skips it. +Closing the reader and leaving the group still run after the wait expires. +This timeout is not a deadline for the entire shutdown. `shutdown()` also ends +an `async for` that waits for a message. In a +`consume_messages()` callback, `shutdown()`, iterating the consumer and a nested +`consume_messages()` raise `RuntimeError`, because the run waits for the +callback: set the shutdown event instead. ```python from apache_iggy import IggyClient diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 4cd6416a90..ecfec2919c 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -2198,6 +2198,7 @@ class IggyClient: polling_retry_interval: datetime.timedelta | None = None, init_retries: builtins.int | None = None, init_retry_interval: datetime.timedelta | None = None, + offset_drain_timeout: datetime.timedelta | None = None, allow_replay: builtins.bool = False, ) -> collections.abc.Awaitable[IggyConsumer]: r""" @@ -2213,9 +2214,12 @@ class IggyClient: each reply, and `CommitPolicy.Auto(...)` makes the consumer store counted positions. `prefetch_bytes` (16 MiB by default) and `prefetch_messages` (the larger of 16,000 and the poll count by default) bound the requests in flight and the queued batches. + `offset_drain_timeout` controls how long shutdown waits for pending automatic offset + stores before closing the reader. It defaults to five seconds; zero skips that wait. Call `shutdown()` before dropping the consumer. Returns the consumer or raises `IggyError` on failure. Raises `ValueError` if - `polling_retry_interval` or `init_retry_interval` is negative or zero. + `polling_retry_interval` or `init_retry_interval` is negative or zero, or if + `offset_drain_timeout` is negative. Consumer groups are not available over HTTP: this call raises `FeatureUnavailable`. Use `Consumer.Single(...)` with `poll_messages(...)` or `reader(...)` there. diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index bd4ffbc91b..50beefcf32 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -2092,9 +2092,12 @@ impl IggyClient { /// each reply, and `CommitPolicy.Auto(...)` makes the consumer store counted positions. /// `prefetch_bytes` (16 MiB by default) and `prefetch_messages` (the larger of 16,000 and /// the poll count by default) bound the requests in flight and the queued batches. + /// `offset_drain_timeout` controls how long shutdown waits for pending automatic offset + /// stores before closing the reader. It defaults to five seconds; zero skips that wait. /// Call `shutdown()` before dropping the consumer. /// Returns the consumer or raises `IggyError` on failure. Raises `ValueError` if - /// `polling_retry_interval` or `init_retry_interval` is negative or zero. + /// `polling_retry_interval` or `init_retry_interval` is negative or zero, or if + /// `offset_drain_timeout` is negative. /// /// Consumer groups are not available over HTTP: this call raises `FeatureUnavailable`. /// Use `Consumer.Single(...)` with `poll_messages(...)` or `reader(...)` there. @@ -2112,6 +2115,7 @@ impl IggyClient { polling_retry_interval=None, init_retries=None, init_retry_interval=None, + offset_drain_timeout=None, allow_replay=false, ))] #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[IggyConsumer]", imports=("collections.abc")))] @@ -2137,6 +2141,8 @@ impl IggyClient { #[gen_stub(override_type(type_repr = "builtins.int | None"))] init_retries: Option, #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] init_retry_interval: Option>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + offset_drain_timeout: Option>, allow_replay: bool, ) -> PyResult> { let mut builder = self @@ -2149,6 +2155,10 @@ impl IggyClient { } else { builder.do_not_create_consumer_group_if_not_exists() }; + if let Some(offset_drain_timeout) = offset_drain_timeout { + builder = + builder.offset_drain_timeout(py_delta_to_iggy_duration(&offset_drain_timeout)?); + } let settings = ConsumerSettings { polling_strategy: polling_strategy.map(|strategy| strategy.inner), poll_options, diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py index b2abd7ccae..ee9ae304b7 100644 --- a/foreign/python/tests/test_consumer_group.py +++ b/foreign/python/tests/test_consumer_group.py @@ -2114,8 +2114,19 @@ async def test_consumer_group_before_login_fails(self, unique_name): ) @pytest.mark.asyncio - async def test_shutdown(self, iggy_client: IggyClient, unique_name): - """Test that the consumer group can be signaled to shutdown.""" + @pytest.mark.parametrize( + "timeout_options", + [ + pytest.param({}, id="omitted"), + pytest.param({"offset_drain_timeout": None}, id="none"), + pytest.param({"offset_drain_timeout": timedelta(seconds=2)}, id="custom"), + pytest.param({"offset_drain_timeout": timedelta(0)}, id="zero"), + ], + ) + async def test_shutdown( + self, iggy_client: IggyClient, unique_name, timeout_options + ): + """Test shutdown accepts the default and configured offset drain timeout.""" consumer_name = unique_name() stream_name = unique_name() topic_name = unique_name() @@ -2130,6 +2141,7 @@ async def test_shutdown(self, iggy_client: IggyClient, unique_name): topic_name, polling_strategy=PollingStrategy.Next(), poll_options=INTERVAL_COMMIT_OPTIONS, + **timeout_options, ) async def take(_: ReceiveMessage) -> None: @@ -2138,6 +2150,26 @@ async def take(_: ReceiveMessage) -> None: shutdown_event.set() await consumer.consume_messages(take, shutdown_event) + await consumer.shutdown() + + group = await iggy_client.get_consumer_group( + stream_name, topic_name, consumer_name + ) + assert group is not None + assert group.members_count == 0 + + @pytest.mark.asyncio + async def test_negative_offset_drain_timeout_fails( + self, iggy_client: IggyClient, unique_name + ): + """Test a negative offset drain timeout is rejected before initialization.""" + with pytest.raises(ValueError, match="duration must not be negative"): + await iggy_client.consumer_group( + unique_name(), + unique_name(), + unique_name(), + offset_drain_timeout=timedelta(microseconds=-1), + ) @pytest.mark.asyncio async def test_shutdown_with_on_read_moves_the_offset_back(