You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2026/07/03 10:03:19 UTC
(iggy) branch master updated: feat(python): expose consumer group create and get APIs (#3581)
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski 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 8c3c99d37 feat(python): expose consumer group create and get APIs (#3581)
8c3c99d37 is described below
commit 8c3c99d37b4b90225d02cb9f0decd93a2297f8f5
Author: WaterWhisperer <wa...@qq.com>
AuthorDate: Fri Jul 3 18:03:15 2026 +0800
feat(python): expose consumer group create and get APIs (#3581)
Closes #3522
---
.../actions/python-maturin/pre-merge/action.yml | 3 +
core/sdk/src/prelude.rs | 25 +-
foreign/python/apache_iggy.pyi | 134 +++++++
foreign/python/src/client.rs | 110 +++++-
foreign/python/src/consumer.rs | 136 ++++++-
foreign/python/src/lib.rs | 8 +-
foreign/python/tests/test_consumer_group.py | 404 ++++++++++++++++++++-
7 files changed, 803 insertions(+), 17 deletions(-)
diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml
index 25409e5ac..3e426c5fb 100644
--- a/.github/actions/python-maturin/pre-merge/action.yml
+++ b/.github/actions/python-maturin/pre-merge/action.yml
@@ -79,6 +79,9 @@ runs:
echo "Running ruff format --check..."
uv run ruff format --check $DIRS_TO_CHECK
+ echo "Running cargo fmt for Python extension..."
+ cargo fmt --manifest-path Cargo.toml -- --check
+
echo "Running pyrefly on SDK..."
uv run pyrefly check "$DIR_SDK"
echo "pyrefly version: $(uv run pyrefly --version)"
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index 53a6f4a95..ec01c1920 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -50,18 +50,19 @@ pub use crate::websocket::websocket_client::WebSocketClient;
pub use iggy_common::{
Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, CacheMetrics, CacheMetricsKey, ClientError,
ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus,
- CompressionAlgorithm, Consumer, ConsumerGroupDetails, ConsumerKind, EncryptorKind,
- GlobalPermissions, HeaderKey, HeaderKind, HeaderValue, HttpClientConfig,
- HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo, IggyByteSize,
- IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage, IggyMessageHeader,
- IggyMessageHeaderView, IggyMessageView, IggyMessageViewIterator, IggyTimestamp, MaxTopicSize,
- Partition, Partitioner, Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages,
- PolledMessages, PollingKind, PollingStrategy, QuicClientConfig, QuicClientConfigBuilder,
- QuicClientReconnectionConfig, SendMessages, Sizeable, SnapshotCompression, Stats, Stream,
- StreamDetails, StreamPermissions, SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder,
- TcpClientReconnectionConfig, Topic, TopicDetails, TopicPermissions, TransportEndpoints,
- TransportProtocol, UserId, UserStatus, Validatable, WebSocketClientConfig,
- WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig, defaults, locking,
+ CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember,
+ ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, HeaderValue,
+ HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo,
+ IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage,
+ IggyMessageHeader, IggyMessageHeaderView, IggyMessageView, IggyMessageViewIterator,
+ IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning, Permissions,
+ PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind, PollingStrategy,
+ QuicClientConfig, QuicClientConfigBuilder, QuicClientReconnectionConfig, SendMessages,
+ Sizeable, SnapshotCompression, Stats, Stream, StreamDetails, StreamPermissions,
+ SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig,
+ Topic, TopicDetails, TopicPermissions, TransportEndpoints, TransportProtocol, UserId,
+ UserStatus, Validatable, WebSocketClientConfig, WebSocketClientConfigBuilder,
+ WebSocketClientReconnectionConfig, defaults, locking,
};
pub use iggy_common::{
Client, ClusterClient, ConsumerGroupClient, ConsumerOffsetClient, MessageClient,
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index ba13c53f2..6a9ad26ba 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -28,6 +28,9 @@ __all__ = [
"AutoCommit",
"AutoCommitAfter",
"AutoCommitWhen",
+ "ConsumerGroup",
+ "ConsumerGroupDetails",
+ "ConsumerGroupMember",
"IggyClient",
"IggyConsumer",
"PollingStrategy",
@@ -231,6 +234,75 @@ class AutoCommitWhen:
...
+@typing.final
+class ConsumerGroup:
+ @property
+ def id(self) -> builtins.int:
+ r"""
+ Gets the unique identifier (numeric) of the consumer group.
+ """
+ @property
+ def name(self) -> builtins.str:
+ r"""
+ Gets the name of the consumer group.
+ """
+ @property
+ def partitions_count(self) -> builtins.int:
+ r"""
+ Gets the number of partitions the consumer group is consuming.
+ """
+ @property
+ def members_count(self) -> builtins.int:
+ r"""
+ Gets the number of members in the consumer group.
+ """
+
+@typing.final
+class ConsumerGroupDetails:
+ @property
+ def id(self) -> builtins.int:
+ r"""
+ Gets the unique identifier (numeric) of the consumer group.
+ """
+ @property
+ def name(self) -> builtins.str:
+ r"""
+ Gets the name of the consumer group.
+ """
+ @property
+ def partitions_count(self) -> builtins.int:
+ r"""
+ Gets the number of partitions the consumer group is consuming.
+ """
+ @property
+ def members_count(self) -> builtins.int:
+ r"""
+ Gets the number of members in the consumer group.
+ """
+ @property
+ def members(self) -> builtins.list[ConsumerGroupMember]:
+ r"""
+ Gets the collection of members in the consumer group.
+ """
+
+@typing.final
+class ConsumerGroupMember:
+ @property
+ def id(self) -> builtins.int:
+ r"""
+ Gets the unique identifier (numeric) of the consumer group member.
+ """
+ @property
+ def partitions_count(self) -> builtins.int:
+ r"""
+ Gets the number of partitions the consumer group member is consuming.
+ """
+ @property
+ def partitions(self) -> builtins.list[builtins.int]:
+ r"""
+ Gets the collection of partitions the consumer group member is consuming.
+ """
+
@typing.final
class IggyClient:
r"""
@@ -385,6 +457,68 @@ class IggyClient:
Raises:
PyRuntimeError: If an identifier is invalid or the request fails.
"""
+ def create_consumer_group(
+ self,
+ stream_id: builtins.str | builtins.int,
+ topic_id: builtins.str | builtins.int,
+ name: builtins.str,
+ ) -> collections.abc.Awaitable[None]:
+ r"""
+ Create a consumer group for a stream and topic.
+
+ Args:
+ stream_id: Stream identifier as `str | int`.
+ topic_id: Topic identifier as `str | int`.
+ name: Consumer group name as `str`.
+
+ Returns:
+ An awaitable that resolves to `None` when the consumer group is created.
+
+ Raises:
+ PyValueError: If an identifier is invalid.
+ PyRuntimeError: If the request fails.
+ """
+ def get_consumer_group(
+ self,
+ stream_id: builtins.str | builtins.int,
+ topic_id: builtins.str | builtins.int,
+ group_id: builtins.str | builtins.int,
+ ) -> collections.abc.Awaitable[ConsumerGroupDetails | None]:
+ r"""
+ Retrieve details for a consumer group from the specified 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 `ConsumerGroupDetails` if the consumer group exists,
+ or `None` otherwise.
+
+ Raises:
+ PyValueError: If an identifier is invalid.
+ PyRuntimeError: If the request fails.
+ """
+ def get_consumer_groups(
+ self,
+ stream_id: builtins.str | builtins.int,
+ topic_id: builtins.str | builtins.int,
+ ) -> collections.abc.Awaitable[list[ConsumerGroup]]:
+ r"""
+ Get all consumer groups for the specified stream and topic.
+
+ Args:
+ stream_id: Stream identifier as `str | int`.
+ topic_id: Topic identifier as `str | int`.
+
+ Returns:
+ An awaitable that resolves to `list[ConsumerGroup]`.
+
+ Raises:
+ PyValueError: If an identifier is invalid.
+ PyRuntimeError: If the request fails.
+ """
def send_messages(
self,
stream: builtins.str | builtins.int,
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 85a567d1d..b604f62c6 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -28,7 +28,10 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods};
use std::str::FromStr;
use std::sync::Arc;
-use crate::consumer::{AutoCommit, IggyConsumer, py_delta_to_iggy_duration};
+use crate::consumer::{
+ AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails,
+ IggyConsumer, py_delta_to_iggy_duration,
+};
use crate::identifier::PyIdentifier;
use crate::receive_message::{PollingStrategy, ReceiveMessage};
use crate::send_message::SendMessage;
@@ -407,6 +410,111 @@ impl IggyClient {
})
}
+ /// Create a consumer group for a stream and topic.
+ ///
+ /// Args:
+ /// stream_id: Stream identifier as `str | int`.
+ /// topic_id: Topic identifier as `str | int`.
+ /// name: Consumer group name as `str`.
+ ///
+ /// Returns:
+ /// An awaitable that resolves to `None` when the consumer group is created.
+ ///
+ /// Raises:
+ /// PyValueError: If an identifier is invalid.
+ /// PyRuntimeError: If the request fails.
+ #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+ fn create_consumer_group<'a>(
+ &self,
+ py: Python<'a>,
+ stream_id: PyIdentifier,
+ topic_id: PyIdentifier,
+ name: String,
+ ) -> PyResult<Bound<'a, PyAny>> {
+ let stream_id = Identifier::try_from(stream_id)?;
+ let topic_id = Identifier::try_from(topic_id)?;
+ let inner = self.inner.clone();
+
+ future_into_py(py, async move {
+ inner
+ .create_consumer_group(&stream_id, &topic_id, &name)
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+ Ok(())
+ })
+ }
+
+ /// Retrieve details for a consumer group from the specified 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 `ConsumerGroupDetails` if the consumer group exists,
+ /// or `None` otherwise.
+ ///
+ /// Raises:
+ /// PyValueError: If an identifier is invalid.
+ /// PyRuntimeError: If the request fails.
+ #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[ConsumerGroupDetails | None]", imports=("collections.abc")))]
+ fn get_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 {
+ let group = inner
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+ Ok(group.map(PyConsumerGroupDetails::from))
+ })
+ }
+
+ /// Get all consumer groups for the specified stream and topic.
+ ///
+ /// Args:
+ /// stream_id: Stream identifier as `str | int`.
+ /// topic_id: Topic identifier as `str | int`.
+ ///
+ /// Returns:
+ /// An awaitable that resolves to `list[ConsumerGroup]`.
+ ///
+ /// Raises:
+ /// PyValueError: If an identifier is invalid.
+ /// PyRuntimeError: If the request fails.
+ #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[ConsumerGroup]]", imports=("collections.abc")))]
+ fn get_consumer_groups<'a>(
+ &self,
+ py: Python<'a>,
+ stream_id: PyIdentifier,
+ topic_id: PyIdentifier,
+ ) -> PyResult<Bound<'a, PyAny>> {
+ let stream_id = Identifier::try_from(stream_id)?;
+ let topic_id = Identifier::try_from(topic_id)?;
+ let inner = self.inner.clone();
+
+ future_into_py(py, async move {
+ let groups = inner
+ .get_consumer_groups(&stream_id, &topic_id)
+ .await
+ .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
+ Ok(groups
+ .into_iter()
+ .map(PyConsumerGroup::from)
+ .collect::<Vec<_>>())
+ })
+ }
+
/// 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/src/consumer.rs b/foreign/python/src/consumer.rs
index ff56de54a..0b6066b68 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -22,9 +22,11 @@ use futures::StreamExt;
use iggy::consumer_ext::{IggyConsumerMessageExt, MessageConsumer};
use iggy::prelude::{
AutoCommit as RustAutoCommit, AutoCommitAfter as RustAutoCommitAfter,
- AutoCommitWhen as RustAutoCommitWhen, *,
+ AutoCommitWhen as RustAutoCommitWhen, ConsumerGroup as RustConsumerGroup,
+ ConsumerGroupDetails as RustConsumerGroupDetails,
+ ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as RustIggyConsumer, IggyDuration,
+ IggyError, ReceivedMessage,
};
-use iggy::prelude::{IggyConsumer as RustIggyConsumer, IggyError, ReceivedMessage};
use pyo3::exceptions::PyStopAsyncIteration;
use pyo3::types::{PyDelta, PyDeltaAccess};
@@ -216,6 +218,136 @@ impl IggyConsumer {
}
}
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct ConsumerGroup {
+ pub(crate) inner: RustConsumerGroup,
+}
+
+impl From<RustConsumerGroup> for ConsumerGroup {
+ fn from(group: RustConsumerGroup) -> Self {
+ Self { inner: group }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl ConsumerGroup {
+ /// Gets the unique identifier (numeric) of the consumer group.
+ #[getter]
+ pub fn id(&self) -> u32 {
+ self.inner.id
+ }
+
+ /// Gets the name of the consumer group.
+ #[getter]
+ pub fn name(&self) -> String {
+ self.inner.name.to_string()
+ }
+
+ /// Gets the number of partitions the consumer group is consuming.
+ #[getter]
+ pub fn partitions_count(&self) -> u32 {
+ self.inner.partitions_count
+ }
+
+ /// Gets the number of members in the consumer group.
+ #[getter]
+ pub fn members_count(&self) -> u32 {
+ self.inner.members_count
+ }
+}
+
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct ConsumerGroupDetails {
+ pub(crate) inner: RustConsumerGroupDetails,
+}
+
+impl From<RustConsumerGroupDetails> for ConsumerGroupDetails {
+ fn from(group: RustConsumerGroupDetails) -> Self {
+ Self { inner: group }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl ConsumerGroupDetails {
+ /// Gets the unique identifier (numeric) of the consumer group.
+ #[getter]
+ pub fn id(&self) -> u32 {
+ self.inner.id
+ }
+
+ /// Gets the name of the consumer group.
+ #[getter]
+ pub fn name(&self) -> String {
+ self.inner.name.to_string()
+ }
+
+ /// Gets the number of partitions the consumer group is consuming.
+ #[getter]
+ pub fn partitions_count(&self) -> u32 {
+ self.inner.partitions_count
+ }
+
+ /// Gets the number of members in the consumer group.
+ #[getter]
+ pub fn members_count(&self) -> u32 {
+ self.inner.members_count
+ }
+
+ /// Gets the collection of members in the consumer group.
+ #[getter]
+ pub fn members(&self) -> Vec<ConsumerGroupMember> {
+ self.inner
+ .members
+ .iter()
+ .map(ConsumerGroupMember::from)
+ .collect()
+ }
+}
+
+#[gen_stub_pyclass]
+#[pyclass]
+pub struct ConsumerGroupMember {
+ pub(crate) inner: RustConsumerGroupMember,
+}
+
+impl From<&RustConsumerGroupMember> for ConsumerGroupMember {
+ fn from(member: &RustConsumerGroupMember) -> Self {
+ Self {
+ inner: RustConsumerGroupMember {
+ id: member.id,
+ partitions_count: member.partitions_count,
+ partitions: member.partitions.clone(),
+ },
+ }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl ConsumerGroupMember {
+ /// Gets the unique identifier (numeric) of the consumer group member.
+ #[getter]
+ pub fn id(&self) -> u32 {
+ self.inner.id
+ }
+
+ /// Gets the number of partitions the consumer group member is consuming.
+ #[getter]
+ pub fn partitions_count(&self) -> u32 {
+ self.inner.partitions_count
+ }
+
+ /// Gets the collection of partitions the consumer group member is consuming.
+ #[getter]
+ pub fn partitions(&self) -> Vec<u32> {
+ self.inner.partitions.clone()
+ }
+}
+
#[pyclass]
pub struct ReceiveMessageIterator {
pub(crate) inner: Arc<Mutex<RustIggyConsumer>>,
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 59b62608b..8c7845606 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -24,7 +24,10 @@ mod stream;
mod topic;
use client::IggyClient;
-use consumer::{AutoCommit, AutoCommitAfter, AutoCommitWhen, IggyConsumer, ReceiveMessageIterator};
+use consumer::{
+ AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup, ConsumerGroupDetails,
+ ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator,
+};
use pyo3::prelude::*;
use receive_message::{PollingStrategy, ReceiveMessage};
use send_message::SendMessage;
@@ -40,6 +43,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<StreamDetails>()?;
m.add_class::<Topic>()?;
m.add_class::<TopicDetails>()?;
+ m.add_class::<ConsumerGroup>()?;
+ m.add_class::<ConsumerGroupDetails>()?;
+ m.add_class::<ConsumerGroupMember>()?;
m.add_class::<PollingStrategy>()?;
m.add_class::<IggyConsumer>()?;
m.add_class::<AutoCommit>()?;
diff --git a/foreign/python/tests/test_consumer_group.py b/foreign/python/tests/test_consumer_group.py
index b88db110d..d193a9a64 100644
--- a/foreign/python/tests/test_consumer_group.py
+++ b/foreign/python/tests/test_consumer_group.py
@@ -34,8 +34,410 @@ from apache_iggy import SendMessage as Message
from .utils import get_server_config, wait_for_ping, wait_for_server
+class TestCreateConsumerGroup:
+ """Test consumer group creation via create_consumer_group."""
+
+ @pytest.mark.asyncio
+ async def test_create_and_get_consumer_group(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test consumer group creation and retrieval."""
+ 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=2,
+ )
+ await iggy_client.create_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+
+ group_by_name = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+ assert group_by_name is not None
+ assert group_by_name.name == group_name
+ assert group_by_name.partitions_count == 2
+ assert group_by_name.members_count == 0
+ assert group_by_name.members == []
+
+ @pytest.mark.asyncio
+ async def test_create_consumer_group_accepts_numeric_stream_and_topic_ids(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test create_consumer_group accepts numeric stream and topic ids."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ stream = await iggy_client.get_stream(stream_name)
+ assert stream is not None
+ stream_id = stream.id
+ 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
+ topic_id = topic.id
+ 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 group.name == group_name
+ assert group.partitions_count == 1
+ assert group.members_count == 0
+ assert group.members == []
+
+ @pytest.mark.asyncio
+ async def test_duplicate_consumer_group_creation_fails(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test create_consumer_group rejects duplicate group names."""
+ 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,
+ )
+
+ with pytest.raises(RuntimeError):
+ await iggy_client.create_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+
+ @pytest.mark.asyncio
+ async def test_create_consumer_group_requires_connection_and_auth(
+ self, unique_name
+ ):
+ """Test create_consumer_group fails before connecting and before login."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(f"{host}:{port}")
+ with pytest.raises(RuntimeError):
+ await client.create_consumer_group(
+ unique_name(),
+ unique_name(),
+ unique_name(),
+ )
+
+ await client.connect()
+ with pytest.raises(RuntimeError):
+ await client.create_consumer_group(
+ unique_name(),
+ unique_name(),
+ unique_name(),
+ )
+
+
+class TestGetConsumerGroup:
+ """Test consumer group retrieval via get_consumer_group."""
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_group_by_name_and_id(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test consumer group lookup works by both name and numeric id."""
+ 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,
+ )
+
+ group_by_name = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+ assert group_by_name is not None
+ assert group_by_name.name == group_name
+ assert group_by_name.partitions_count == 1
+ assert group_by_name.members_count == 0
+ assert group_by_name.members == []
+
+ group_by_id = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ group_by_name.id,
+ )
+ assert group_by_id is not None
+ assert group_by_id.id == group_by_name.id
+ assert group_by_id.name == group_name
+ assert group_by_id.partitions_count == 1
+ assert group_by_id.members_count == 0
+ assert group_by_id.members == []
+
+ @pytest.mark.asyncio
+ async def test_get_nonexistent_consumer_group(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test getting a non-existent consumer group by name or numeric id."""
+ 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,
+ )
+
+ group_by_name = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ unique_name(),
+ )
+ assert group_by_name is None
+
+ group_by_id = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ 999999,
+ )
+ assert group_by_id is None
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_group_returns_members(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test get_consumer_group exposes non-empty member details."""
+ consumer_name = unique_name()
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ await iggy_client.consumer_group(
+ consumer_name,
+ stream_name,
+ topic_name,
+ partition_id,
+ PollingStrategy.Next(),
+ 10,
+ )
+
+ group = await iggy_client.get_consumer_group(
+ stream_name,
+ topic_name,
+ consumer_name,
+ )
+ assert group is not None
+ assert group.name == consumer_name
+ assert group.partitions_count == 1
+ assert group.members_count == 1
+
+ members = group.members
+ assert len(members) == 1
+ assert members[0].partitions_count == 1
+ assert members[0].partitions == [partition_id]
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_group_requires_connection_and_auth(self, unique_name):
+ """Test get_consumer_group fails before connecting and before login."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(f"{host}:{port}")
+ with pytest.raises(RuntimeError):
+ await client.get_consumer_group(
+ unique_name(),
+ unique_name(),
+ unique_name(),
+ )
+
+ await client.connect()
+ with pytest.raises(RuntimeError):
+ await client.get_consumer_group(
+ unique_name(),
+ unique_name(),
+ unique_name(),
+ )
+
+
+class TestGetConsumerGroups:
+ """Test listing consumer groups via get_consumer_groups."""
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_groups_in_empty_topic_returns_empty_list(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test get_consumer_groups returns an empty list for a topic with no groups."""
+ 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,
+ )
+
+ groups = await iggy_client.get_consumer_groups(stream_name, topic_name)
+ assert groups == []
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_groups_lists_topic_groups(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test listing consumer groups returns groups for the requested topic."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ first_group_name = unique_name()
+ second_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,
+ first_group_name,
+ )
+ await iggy_client.create_consumer_group(
+ stream_name,
+ topic_name,
+ second_group_name,
+ )
+
+ groups = await iggy_client.get_consumer_groups(stream_name, topic_name)
+
+ assert len(groups) == 2
+ assert [group.name for group in groups] == [first_group_name, second_group_name]
+ assert all(group.partitions_count == 1 for group in groups)
+ assert all(group.members_count == 0 for group in groups)
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_groups_accepts_numeric_stream_and_topic_ids(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test get_consumer_groups accepts numeric stream and topic ids."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+
+ await iggy_client.create_stream(stream_name)
+ stream = await iggy_client.get_stream(stream_name)
+ assert stream is not None
+ stream_id = stream.id
+ 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
+ topic_id = topic.id
+ assert isinstance(topic_id, int)
+ await iggy_client.create_consumer_group(stream_id, topic_id, group_name)
+
+ groups = await iggy_client.get_consumer_groups(stream_id, topic_id)
+
+ assert len(groups) == 1
+ assert groups[0].name == group_name
+ assert groups[0].partitions_count == 1
+ assert groups[0].members_count == 0
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_groups_returns_same_result_when_called_repeatedly(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test repeated get_consumer_groups calls return a stable topic view."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_names = [unique_name() for _ in range(2)]
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+ for group_name in group_names:
+ await iggy_client.create_consumer_group(
+ stream_name,
+ topic_name,
+ group_name,
+ )
+
+ first = await iggy_client.get_consumer_groups(stream_name, topic_name)
+ second = await iggy_client.get_consumer_groups(stream_name, topic_name)
+
+ assert [group.name for group in first] == group_names
+ assert [group.id for group in first] == [group.id for group in second]
+ assert [group.name for group in first] == [group.name for group in second]
+
+ @pytest.mark.asyncio
+ async def test_get_consumer_groups_requires_connection_and_auth(self, unique_name):
+ """Test get_consumer_groups fails before connecting and before login."""
+ host, port = get_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(f"{host}:{port}")
+ with pytest.raises(RuntimeError):
+ await client.get_consumer_groups(unique_name(), unique_name())
+
+ await client.connect()
+ with pytest.raises(RuntimeError):
+ await client.get_consumer_groups(unique_name(), unique_name())
+
+
class TestConsumerGroup:
- """Test consumer groups."""
+ """Test consumer group consumers."""
@pytest.mark.asyncio
async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):