feat(python): expose consumer shutdown and offset drain timeout

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

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.

foreign/python/src/client.rs

@@ -2085,9 +2085,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.
@@ -2105,6 +2108,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")))]
@@ -2130,6 +2134,8 @@ impl IggyClient {
         #[gen_stub(override_type(type_repr = "builtins.int | None"))] init_retries: Option<u32>,
         #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
         init_retry_interval: Option<Py<PyDelta>>,
+        #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+        offset_drain_timeout: Option<Py<PyDelta>>,
         allow_replay: bool,
     ) -> PyResult<Bound<'a, PyAny>> {
         let mut builder = self
@@ -2142,6 +2148,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,

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(