This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 7b79dbdc4 feat(python): expose consumer group delete join and leave
APIs (#3607)
7b79dbdc4 is described below
commit 7b79dbdc4e79edb76cdcc9b8476ad7dd5d17433c
Author: WaterWhisperer <[email protected]>
AuthorDate: Mon Jul 13 12:49:59 2026 +0800
feat(python): expose consumer group delete join and leave APIs (#3607)
---
core/integration/tests/sdk/consumer_group.rs | 163 ++++++++++++
core/integration/tests/sdk/mod.rs | 1 +
core/sdk/src/clients/consumer.rs | 29 ++
foreign/python/apache_iggy.pyi | 71 +++++
foreign/python/src/client.rs | 113 ++++++++
foreign/python/tests/test_consumer_group.py | 384 +++++++++++++++++++++++++++
6 files changed, 761 insertions(+)
diff --git a/core/integration/tests/sdk/consumer_group.rs
b/core/integration/tests/sdk/consumer_group.rs
new file mode 100644
index 000000000..5ea300c6b
--- /dev/null
+++ b/core/integration/tests/sdk/consumer_group.rs
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::str::FromStr;
+use std::time::Duration;
+
+use futures::StreamExt;
+use iggy::prelude::*;
+use integration::iggy_harness;
+use tokio::time::timeout;
+
+const STREAM_NAME: &str = "consumer-group-rejoin-stream";
+const TOPIC_NAME: &str = "consumer-group-rejoin-topic";
+const CONSUMER_GROUP_NAME: &str = "consumer-group-rejoin-group";
+const CONSUMER_USERNAME: &str = "consumer-group-rejoin-user";
+const CONSUMER_PASSWORD: &str = "password123";
+const CONSUMER_REJOIN_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
+async fn consumer_group_retries_rejoin_after_failure(harness: &TestHarness) {
+ let root_client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
+ let stream_id = Identifier::named(STREAM_NAME).unwrap();
+ let topic_id = Identifier::named(TOPIC_NAME).unwrap();
+ let group_id = Identifier::named(CONSUMER_GROUP_NAME).unwrap();
+ let user_id = Identifier::named(CONSUMER_USERNAME).unwrap();
+ let consumer_permissions = Permissions {
+ global: GlobalPermissions {
+ read_streams: true,
+ ..Default::default()
+ },
+ streams: None,
+ };
+
+ root_client.create_stream(STREAM_NAME).await.unwrap();
+ root_client
+ .create_topic(
+ &stream_id,
+ TOPIC_NAME,
+ 1,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+ root_client
+ .create_consumer_group(&stream_id, &topic_id, CONSUMER_GROUP_NAME)
+ .await
+ .unwrap();
+ root_client
+ .create_user(
+ CONSUMER_USERNAME,
+ CONSUMER_PASSWORD,
+ UserStatus::Active,
+ Some(consumer_permissions.clone()),
+ )
+ .await
+ .unwrap();
+
+ let consumer_client = harness.new_client().await.expect("Failed to create
client");
+ consumer_client
+ .login_user(CONSUMER_USERNAME, CONSUMER_PASSWORD)
+ .await
+ .unwrap();
+
+ let mut consumer = consumer_client
+ .consumer_group(CONSUMER_GROUP_NAME, STREAM_NAME, TOPIC_NAME)
+ .unwrap()
+ .batch_length(1)
+ .auto_join_consumer_group()
+ .build();
+ consumer.init().await.unwrap();
+
+ let group = root_client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap()
+ .expect("Consumer group should exist");
+ assert_eq!(group.members_count, 1);
+ assert_eq!(group.members.len(), 1);
+
+ let mut messages = vec![IggyMessage::from_str("message").unwrap()];
+ root_client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+
+ consumer_client
+ .leave_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap();
+
+ let group = root_client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap()
+ .expect("Consumer group should exist");
+ assert_eq!(group.members_count, 0);
+ assert!(group.members.is_empty());
+
+ root_client
+ .update_permissions(
+ &user_id,
+ Some(Permissions {
+ global: GlobalPermissions {
+ poll_messages: true,
+ ..Default::default()
+ },
+ streams: None,
+ }),
+ )
+ .await
+ .unwrap();
+
+ let rejoin_result = timeout(CONSUMER_REJOIN_TIMEOUT, consumer.next())
+ .await
+ .expect("Consumer rejoin should fail before timeout")
+ .expect("Consumer stream should remain open");
+ assert!(matches!(rejoin_result, Err(IggyError::Unauthorized)));
+
+ root_client
+ .update_permissions(&user_id, Some(consumer_permissions))
+ .await
+ .unwrap();
+
+ let received = timeout(CONSUMER_REJOIN_TIMEOUT, consumer.next())
+ .await
+ .expect("Consumer should recover before timeout")
+ .expect("Consumer stream should remain open")
+ .expect("Consumer should rejoin after its membership is revoked");
+ assert_eq!(received.message.payload, "message");
+
+ let group = root_client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap()
+ .expect("Consumer group should exist");
+ assert_eq!(group.members_count, 1);
+ assert_eq!(group.members.len(), 1);
+}
diff --git a/core/integration/tests/sdk/mod.rs
b/core/integration/tests/sdk/mod.rs
index 0cd484a18..6c4dabe19 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+mod consumer_group;
mod hello_world;
#[cfg(feature = "vsr")]
mod http_refresh;
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 918245953..21b98fa4f 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -658,6 +658,8 @@ impl IggyConsumer {
let last_consumed_offset = self.last_consumed_offsets.clone();
let allow_replay = self.allow_replay;
let is_consumer_group = self.is_consumer_group;
+ let auto_join_consumer_group = self.auto_join_consumer_group;
+ let create_consumer_group_if_not_exists =
self.create_consumer_group_if_not_exists;
let joined_consumer_group = self.joined_consumer_group.clone();
async move {
@@ -783,6 +785,33 @@ impl IggyConsumer {
let error = polled_messages.unwrap_err();
error!("Failed to poll messages: {error}");
+ if is_consumer_group
+ && auto_join_consumer_group
+ && matches!(&error, IggyError::ConsumerGroupMemberNotFound(..))
+ {
+ joined_consumer_group.store(false, ORDERING);
+ let consumer_name = consumer.id.as_string();
+ info!(
+ "Consumer group membership was revoked for consumer:
{consumer_name}, stream: {stream_id}, topic: {topic_id}. Rejoining..."
+ );
+ if let Err(error) = Self::initialize_consumer_group(
+ client,
+ create_consumer_group_if_not_exists,
+ stream_id,
+ topic_id,
+ consumer,
+ &consumer_name,
+ joined_consumer_group.clone(),
+ )
+ .await
+ {
+ // Allow the next poll to retry rejoining
+ joined_consumer_group.store(true, ORDERING);
+ return Err(error);
+ }
+ return Ok(PolledMessages::empty());
+ }
+
// Handle connection/auth errors - disable polling until event
task re-enables
// it after reconnection and rejoin complete
if matches!(
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 6a9ad26ba..5b5366938 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -519,6 +519,77 @@ class IggyClient:
PyValueError: If an identifier is invalid.
PyRuntimeError: If the request fails.
"""
+ def delete_consumer_group(
+ self,
+ stream_id: builtins.str | builtins.int,
+ topic_id: builtins.str | builtins.int,
+ group_id: builtins.str | builtins.int,
+ ) -> collections.abc.Awaitable[None]:
+ r"""
+ Delete a consumer group for a stream and topic.
+
+ Args:
+ stream_id: Stream identifier as `str | int`.
+ topic_id: Topic identifier as `str | int`.
+ group_id: Consumer group identifier as `str | int`.
+
+ Returns:
+ An awaitable that resolves to `None` when the consumer group is
deleted.
+
+ Raises:
+ PyValueError: If a string identifier is invalid.
+ PyRuntimeError: If the request fails.
+ """
+ def join_consumer_group(
+ self,
+ stream_id: builtins.str | builtins.int,
+ topic_id: builtins.str | builtins.int,
+ group_id: builtins.str | builtins.int,
+ ) -> collections.abc.Awaitable[None]:
+ r"""
+ Join a consumer group for a stream and topic.
+
+ This method only registers the current client as a group member. To
consume messages
+ as a group, use `consumer_group()`, which enables auto-join by default.
+
+ Args:
+ stream_id: Stream identifier as `str | int`.
+ topic_id: Topic identifier as `str | int`.
+ group_id: Consumer group identifier as `str | int`.
+
+ Returns:
+ An awaitable that resolves to `None` when the client joins the
consumer group.
+
+ Raises:
+ PyValueError: If a string identifier is invalid.
+ PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+ """
+ def leave_consumer_group(
+ self,
+ stream_id: builtins.str | builtins.int,
+ topic_id: builtins.str | builtins.int,
+ group_id: builtins.str | builtins.int,
+ ) -> collections.abc.Awaitable[None]:
+ r"""
+ Leave a consumer group for a stream and topic.
+
+ Args:
+ stream_id: Stream identifier as `str | int`.
+ topic_id: Topic identifier as `str | int`.
+ group_id: Consumer group identifier as `str | int`.
+
+ Returns:
+ An awaitable that resolves to `None` when the client leaves the
consumer group.
+
+ Note:
+ Consumers created from this client for the same group share one
server-side
+ membership. Leaving revokes that membership. Consumers with
auto-join enabled
+ rejoin on their next poll.
+
+ Raises:
+ PyValueError: If a string identifier is invalid.
+ PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+ """
def send_messages(
self,
stream: builtins.str | builtins.int,
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index b604f62c6..2a36ef333 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -515,6 +515,119 @@ impl IggyClient {
})
}
+ /// Delete a consumer group for a stream and topic.
+ ///
+ /// Args:
+ /// stream_id: Stream identifier as `str | int`.
+ /// topic_id: Topic identifier as `str | int`.
+ /// group_id: Consumer group identifier as `str | int`.
+ ///
+ /// Returns:
+ /// An awaitable that resolves to `None` when the consumer group is
deleted.
+ ///
+ /// Raises:
+ /// PyValueError: If a string identifier is invalid.
+ /// PyRuntimeError: If the request fails.
+
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
+ fn delete_consumer_group<'a>(
+ &self,
+ py: Python<'a>,
+ stream_id: PyIdentifier,
+ topic_id: PyIdentifier,
+ group_id: PyIdentifier,
+ ) -> PyResult<Bound<'a, PyAny>> {
+ let stream_id = Identifier::try_from(stream_id)?;
+ let topic_id = Identifier::try_from(topic_id)?;
+ let group_id = Identifier::try_from(group_id)?;
+ let inner = self.inner.clone();
+
+ future_into_py(py, async move {
+ inner
+ .delete_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?;
+ Ok(())
+ })
+ }
+
+ /// Join a consumer group for a stream and topic.
+ ///
+ /// This method only registers the current client as a group member. To
consume messages
+ /// as a group, use `consumer_group()`, which enables auto-join by default.
+ ///
+ /// Args:
+ /// stream_id: Stream identifier as `str | int`.
+ /// topic_id: Topic identifier as `str | int`.
+ /// group_id: Consumer group identifier as `str | int`.
+ ///
+ /// Returns:
+ /// An awaitable that resolves to `None` when the client joins the
consumer group.
+ ///
+ /// Raises:
+ /// PyValueError: If a string identifier is invalid.
+ /// PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
+ fn join_consumer_group<'a>(
+ &self,
+ py: Python<'a>,
+ stream_id: PyIdentifier,
+ topic_id: PyIdentifier,
+ group_id: PyIdentifier,
+ ) -> PyResult<Bound<'a, PyAny>> {
+ let stream_id = Identifier::try_from(stream_id)?;
+ let topic_id = Identifier::try_from(topic_id)?;
+ let group_id = Identifier::try_from(group_id)?;
+ let inner = self.inner.clone();
+
+ future_into_py(py, async move {
+ inner
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?;
+ Ok(())
+ })
+ }
+
+ /// Leave a consumer group for a stream and topic.
+ ///
+ /// Args:
+ /// stream_id: Stream identifier as `str | int`.
+ /// topic_id: Topic identifier as `str | int`.
+ /// group_id: Consumer group identifier as `str | int`.
+ ///
+ /// Returns:
+ /// An awaitable that resolves to `None` when the client leaves the
consumer group.
+ ///
+ /// Note:
+ /// Consumers created from this client for the same group share one
server-side
+ /// membership. Leaving revokes that membership. Consumers with
auto-join enabled
+ /// rejoin on their next poll.
+ ///
+ /// Raises:
+ /// PyValueError: If a string identifier is invalid.
+ /// PyRuntimeError: If the request fails, including `Feature is
unavailable` on HTTP transport.
+
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
+ fn leave_consumer_group<'a>(
+ &self,
+ py: Python<'a>,
+ stream_id: PyIdentifier,
+ topic_id: PyIdentifier,
+ group_id: PyIdentifier,
+ ) -> PyResult<Bound<'a, PyAny>> {
+ let stream_id = Identifier::try_from(stream_id)?;
+ let topic_id = Identifier::try_from(topic_id)?;
+ let group_id = Identifier::try_from(group_id)?;
+ let inner = self.inner.clone();
+
+ future_into_py(py, async move {
+ inner
+ .leave_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?;
+ Ok(())
+ })
+ }
+
/// Sends a list of messages to the specified topic.
/// Returns Ok(()) on successful sending or a PyRuntimeError on failure.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
diff --git a/foreign/python/tests/test_consumer_group.py
b/foreign/python/tests/test_consumer_group.py
index d193a9a64..acb126048 100644
--- a/foreign/python/tests/test_consumer_group.py
+++ b/foreign/python/tests/test_consumer_group.py
@@ -34,6 +34,36 @@ from apache_iggy import SendMessage as Message
from .utils import get_server_config, wait_for_ping, wait_for_server
+async def _create_consumer_group_with_numeric_ids(
+ iggy_client: IggyClient,
+ unique_name,
+ group_name: str,
+) -> tuple[int, int, int]:
+ stream_name = unique_name()
+ topic_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ stream = await iggy_client.get_stream(stream_name)
+ assert stream is not None
+ assert isinstance(stream.id, int)
+
+ await iggy_client.create_topic(
+ stream=stream.id,
+ name=topic_name,
+ partitions_count=1,
+ )
+ topic = await iggy_client.get_topic(stream.id, topic_name)
+ assert topic is not None
+ assert isinstance(topic.id, int)
+
+ await iggy_client.create_consumer_group(stream.id, topic.id, group_name)
+ group = await iggy_client.get_consumer_group(stream.id, topic.id,
group_name)
+ assert group is not None
+ assert isinstance(group.id, int)
+
+ return stream.id, topic.id, group.id
+
+
class TestCreateConsumerGroup:
"""Test consumer group creation via create_consumer_group."""
@@ -436,6 +466,360 @@ class TestGetConsumerGroups:
await client.get_consumer_groups(unique_name(), unique_name())
+class TestDeleteConsumerGroup:
+ """Test deleting consumer groups via delete_consumer_group."""
+
+ @pytest.mark.asyncio
+ async def test_delete_consumer_group_removes_only_target_group(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test delete_consumer_group removes the target group only."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_to_delete = unique_name()
+ group_to_keep = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+ await iggy_client.create_consumer_group(
+ stream_name,
+ topic_name,
+ group_to_delete,
+ )
+ await iggy_client.create_consumer_group(
+ stream_name,
+ topic_name,
+ group_to_keep,
+ )
+
+ await iggy_client.delete_consumer_group(
+ stream_name,
+ topic_name,
+ group_to_delete,
+ )
+
+ assert (
+ await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ group_to_delete,
+ )
+ is None
+ )
+
+ remaining = await iggy_client.get_consumer_groups(stream_name,
topic_name)
+ assert len(remaining) == 1
+ assert remaining[0].name == group_to_keep
+ assert remaining[0].partitions_count == 1
+ assert remaining[0].members_count == 0
+
+ @pytest.mark.asyncio
+ async def test_delete_consumer_group_accepts_numeric_ids(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test delete_consumer_group accepts numeric stream, topic, and group
ids."""
+ group_name = unique_name()
+
+ stream_id, topic_id, group_id = await
_create_consumer_group_with_numeric_ids(
+ iggy_client,
+ unique_name,
+ group_name,
+ )
+
+ await iggy_client.delete_consumer_group(stream_id, topic_id, group_id)
+
+ assert (
+ await iggy_client.get_consumer_group(stream_id, topic_id,
group_id) is None
+ )
+
+ @pytest.mark.asyncio
+ async def test_delete_nonexistent_consumer_group_fails(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test delete_consumer_group raises for a non-existent group."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ with pytest.raises(RuntimeError, match="Consumer group with ID:"):
+ await iggy_client.delete_consumer_group(
+ stream_name,
+ topic_name,
+ unique_name(),
+ )
+
+ @pytest.mark.asyncio
+ async def test_delete_consumer_group_removes_live_member(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test deleting a consumer group also removes its live member."""
+ group_name = unique_name()
+
+ stream_id, topic_id, group_id = await
_create_consumer_group_with_numeric_ids(
+ iggy_client,
+ unique_name,
+ group_name,
+ )
+ await iggy_client.join_consumer_group(stream_id, topic_id, group_id)
+
+ group = await iggy_client.get_consumer_group(
+ stream_id,
+ topic_id,
+ group_id,
+ )
+ assert group is not None
+ assert group.members_count == 1
+ assert len(group.members) == 1
+
+ await iggy_client.delete_consumer_group(stream_id, topic_id, group_id)
+
+ assert (
+ await iggy_client.get_consumer_group(
+ stream_id,
+ topic_id,
+ group_id,
+ )
+ is None
+ )
+ assert await iggy_client.get_consumer_groups(stream_id, topic_id) == []
+
+
+class TestJoinConsumerGroup:
+ """Test joining consumer groups via join_consumer_group."""
+
+ @pytest.mark.asyncio
+ async def test_join_consumer_group_adds_member_once_when_repeated(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test join_consumer_group adds one stable member for repeated
joins."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+ await iggy_client.create_consumer_group(stream_name, topic_name,
group_name)
+
+ await iggy_client.join_consumer_group(stream_name, topic_name,
group_name)
+ await iggy_client.join_consumer_group(stream_name, topic_name,
group_name)
+
+ group = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+ assert group is not None
+ assert group.name == group_name
+ assert group.partitions_count == 1
+ assert group.members_count == 1
+ assert len(group.members) == 1
+ assert group.members[0].partitions_count == 1
+ assert group.members[0].partitions == [0]
+
+ @pytest.mark.asyncio
+ async def test_join_consumer_group_accepts_numeric_ids(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test join_consumer_group accepts numeric stream, topic, and group
ids."""
+ group_name = unique_name()
+
+ stream_id, topic_id, group_id = await
_create_consumer_group_with_numeric_ids(
+ iggy_client,
+ unique_name,
+ group_name,
+ )
+
+ await iggy_client.join_consumer_group(stream_id, topic_id, group_id)
+
+ joined_group = await iggy_client.get_consumer_group(
+ stream_id,
+ topic_id,
+ group_id,
+ )
+ assert joined_group is not None
+ assert joined_group.id == group_id
+ assert joined_group.name == group_name
+ assert joined_group.partitions_count == 1
+ assert joined_group.members_count == 1
+ assert len(joined_group.members) == 1
+ assert joined_group.members[0].partitions_count == 1
+ assert joined_group.members[0].partitions == [0]
+
+ @pytest.mark.asyncio
+ async def test_join_nonexistent_consumer_group_fails(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test join_consumer_group raises for a non-existent group."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ with pytest.raises(RuntimeError, match="Consumer group with ID:"):
+ await iggy_client.join_consumer_group(
+ stream_name,
+ topic_name,
+ unique_name(),
+ )
+
+
+class TestLeaveConsumerGroup:
+ """Test leaving consumer groups via leave_consumer_group."""
+
+ @pytest.mark.asyncio
+ async def test_leave_consumer_group_removes_member(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test leave_consumer_group removes the current client from the
group."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+ await iggy_client.create_consumer_group(stream_name, topic_name,
group_name)
+ await iggy_client.join_consumer_group(stream_name, topic_name,
group_name)
+
+ await iggy_client.leave_consumer_group(stream_name, topic_name,
group_name)
+
+ group = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+ assert group is not None
+ assert group.name == group_name
+ assert group.partitions_count == 1
+ assert group.members_count == 0
+ assert group.members == []
+
+ @pytest.mark.asyncio
+ async def test_leave_consumer_group_accepts_numeric_ids(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test leave_consumer_group accepts numeric stream, topic, and group
ids."""
+ group_name = unique_name()
+
+ stream_id, topic_id, group_id = await
_create_consumer_group_with_numeric_ids(
+ iggy_client,
+ unique_name,
+ group_name,
+ )
+ await iggy_client.join_consumer_group(stream_id, topic_id, group_id)
+
+ await iggy_client.leave_consumer_group(stream_id, topic_id, group_id)
+
+ left_group = await iggy_client.get_consumer_group(
+ stream_id,
+ topic_id,
+ group_id,
+ )
+ assert left_group is not None
+ assert left_group.id == group_id
+ assert left_group.name == group_name
+ assert left_group.partitions_count == 1
+ assert left_group.members_count == 0
+ assert left_group.members == []
+
+ @pytest.mark.asyncio
+ async def test_leave_consumer_group_twice_fails_second_time(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test leaving a group twice raises on the second call."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+ await iggy_client.create_consumer_group(stream_name, topic_name,
group_name)
+ await iggy_client.join_consumer_group(stream_name, topic_name,
group_name)
+
+ await iggy_client.leave_consumer_group(stream_name, topic_name,
group_name)
+ with pytest.raises(
+ RuntimeError,
+ match="Consumer group member with client ID:",
+ ):
+ await iggy_client.leave_consumer_group(stream_name, topic_name,
group_name)
+
+ @pytest.mark.asyncio
+ async def test_leave_nonexistent_consumer_group_fails(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test leave_consumer_group raises for a non-existent group."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ with pytest.raises(RuntimeError, match="Consumer group with ID:"):
+ await iggy_client.leave_consumer_group(
+ stream_name,
+ topic_name,
+ unique_name(),
+ )
+
+
[email protected](
+ "method_name",
+ [
+ "delete_consumer_group",
+ "join_consumer_group",
+ "leave_consumer_group",
+ ],
+)
[email protected]
+async def test_consumer_group_lifecycle_requires_connection_and_auth(
+ method_name, unique_name
+):
+ """Test lifecycle methods fail before connecting and before login."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(f"{host}:{port}")
+ method = getattr(client, method_name)
+ identifiers = (unique_name(), unique_name(), unique_name())
+
+ with pytest.raises(RuntimeError, match="Disconnected"):
+ await method(*identifiers)
+
+ await client.connect()
+ with pytest.raises(RuntimeError, match="Unauthenticated"):
+ await method(*identifiers)
+
+
class TestConsumerGroup:
"""Test consumer group consumers."""
