This is an automated email from the ASF dual-hosted git repository.
hubcio 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 35d2c4244 feat(python): let poll_messages take a consumer (#3877)
35d2c4244 is described below
commit 35d2c4244d09d06bab86ecadc30f6e3d391a0b5d
Author: Ethan Lin <[email protected]>
AuthorDate: Thu Aug 20 21:28:51 2026 +0800
feat(python): let poll_messages take a consumer (#3877)
---
bdd/python/tests/test_basic_messaging.py | 3 +-
examples/python/basic/consumer.py | 4 +-
examples/python/getting-started/consumer.py | 3 +
examples/python/message-headers/common.py | 7 +-
.../message-headers/plain-headers/consumer.py | 2 +-
.../message-headers/typed-headers/consumer.py | 2 +-
foreign/python/apache_iggy.pyi | 302 ++++++++++-------
foreign/python/src/client.rs | 17 +-
foreign/python/src/consumer.rs | 31 +-
foreign/python/src/identifier.rs | 20 +-
foreign/python/src/lib.rs | 3 +-
foreign/python/tests/test_message_operations.py | 377 +++++++++++++++++++++
foreign/python/tests/test_tls.py | 3 +-
13 files changed, 620 insertions(+), 154 deletions(-)
diff --git a/bdd/python/tests/test_basic_messaging.py
b/bdd/python/tests/test_basic_messaging.py
index d489ef4e9..cbaf3869a 100644
--- a/bdd/python/tests/test_basic_messaging.py
+++ b/bdd/python/tests/test_basic_messaging.py
@@ -22,7 +22,7 @@ Basic messaging BDD test implementation for Python SDK
import asyncio
import socket
-from apache_iggy import IggyClient, PollingStrategy, SendMessage
+from apache_iggy import Consumer, IggyClient, PollingStrategy, SendMessage
from pytest_bdd import given, parsers, scenarios, then, when
# Load scenarios from the shared feature file
@@ -221,6 +221,7 @@ def poll_messages(context, stream_id, topic_id,
partition_id, start_offset):
context.last_polled_messages = await context.client.poll_messages(
stream=stream_id,
topic=topic_id,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Offset(value=start_offset),
count=100, # Poll up to 100 messages
diff --git a/examples/python/basic/consumer.py
b/examples/python/basic/consumer.py
index e20cf3175..e90df3c19 100644
--- a/examples/python/basic/consumer.py
+++ b/examples/python/basic/consumer.py
@@ -19,7 +19,7 @@ import argparse
import asyncio
from typing import NamedTuple
-from apache_iggy import IggyClient, PollingStrategy, ReceiveMessage
+from apache_iggy import Consumer, IggyClient, PollingStrategy, ReceiveMessage
from loguru import logger
STREAM_NAME = "sample-stream"
@@ -27,6 +27,7 @@ TOPIC_NAME = "sample-topic"
STREAM_ID = 0
TOPIC_ID = 0
PARTITION_ID = 0
+CONSUMER_NAME = "sample-consumer"
BATCHES_LIMIT = 5
@@ -73,6 +74,7 @@ async def consume_messages(client: IggyClient):
polled_messages = await client.poll_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
+ consumer=Consumer.Single(CONSUMER_NAME),
partition_id=PARTITION_ID,
polling_strategy=PollingStrategy.Next(),
count=messages_per_batch,
diff --git a/examples/python/getting-started/consumer.py
b/examples/python/getting-started/consumer.py
index db0a6edb1..651084104 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -23,6 +23,7 @@ from datetime import timedelta
from apache_iggy import (
AutoLogin,
+ Consumer,
IggyClient,
PollingStrategy,
ReceiveMessage,
@@ -36,6 +37,7 @@ TOPIC_NAME = "sample-topic"
STREAM_ID = 0
TOPIC_ID = 0
PARTITION_ID = 0
+CONSUMER_NAME = "sample-consumer"
BATCHES_LIMIT = 5
@@ -150,6 +152,7 @@ async def consume_messages(client: IggyClient):
polled_messages = await client.poll_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
+ consumer=Consumer.Single(CONSUMER_NAME),
partition_id=PARTITION_ID,
polling_strategy=PollingStrategy.Next(),
count=messages_per_batch,
diff --git a/examples/python/message-headers/common.py
b/examples/python/message-headers/common.py
index 967afe930..6f8acef71 100644
--- a/examples/python/message-headers/common.py
+++ b/examples/python/message-headers/common.py
@@ -25,6 +25,7 @@ from dataclasses import dataclass
from enum import Enum
from apache_iggy import (
+ Consumer,
HeaderKey,
HeaderValue,
IggyClient,
@@ -176,11 +177,14 @@ async def produce_messages(client: IggyClient,
build_headers: HeadersBuilder) ->
logger.info(f"Sent {sent_batches} batches of messages, exiting.")
-async def consume_messages(client: IggyClient, handle_message: MessageHandler)
-> None:
+async def consume_messages(
+ client: IggyClient, handle_message: MessageHandler, consumer_name: str
+) -> None:
interval = 0.5
logger.info(
f"Messages will be consumed from stream: {STREAM_NAME}, "
f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} "
+ f"as consumer: {consumer_name} "
f"with interval {interval * 1000} ms."
)
consumed_batches = 0
@@ -191,6 +195,7 @@ async def consume_messages(client: IggyClient,
handle_message: MessageHandler) -
polled_messages = await client.poll_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
+ consumer=Consumer.Single(consumer_name),
partition_id=PARTITION_ID,
polling_strategy=PollingStrategy.Next(),
count=MESSAGES_PER_BATCH,
diff --git a/examples/python/message-headers/plain-headers/consumer.py
b/examples/python/message-headers/plain-headers/consumer.py
index 92874b022..db18d7599 100644
--- a/examples/python/message-headers/plain-headers/consumer.py
+++ b/examples/python/message-headers/plain-headers/consumer.py
@@ -70,7 +70,7 @@ def get_message_type(
async def main() -> None:
args = parse_args()
client = await connect(args.connection_string)
- await consume_messages(client, handle_message)
+ await consume_messages(client, handle_message, "plain-headers-consumer")
if __name__ == "__main__":
diff --git a/examples/python/message-headers/typed-headers/consumer.py
b/examples/python/message-headers/typed-headers/consumer.py
index e24d531b2..60cbffc80 100644
--- a/examples/python/message-headers/typed-headers/consumer.py
+++ b/examples/python/message-headers/typed-headers/consumer.py
@@ -70,7 +70,7 @@ def get_message_type(headers: UserHeaders | None) ->
OrderType | None:
async def main() -> None:
args = parse_args()
client = await connect(args.connection_string)
- await consume_messages(client, handle_message)
+ await consume_messages(client, handle_message, "typed-headers-consumer")
if __name__ == "__main__":
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 50b65786d..1bdd5cafc 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -30,12 +30,13 @@ __all__ = [
"AutoCommitAfter",
"AutoCommitWhen",
"AutoLogin",
+ "Consumer",
"ConsumerGroup",
"ConsumerGroupDetails",
"ConsumerGroupMember",
+ "GlobalPermissions",
"HeaderKey",
"HeaderValue",
- "GlobalPermissions",
"IggyClient",
"IggyConsumer",
"IggyExpiry",
@@ -55,10 +56,10 @@ __all__ = [
"Topic",
"TopicDetails",
"TopicPermissions",
+ "UserHeaders",
"UserInfo",
"UserInfoDetails",
"UserStatus",
- "UserHeaders",
]
class AutoCommit:
@@ -289,6 +290,35 @@ class AutoLogin:
"""
def __repr__(self) -> builtins.str: ...
+class Consumer:
+ r"""
+ The consumer polling the messages. It selects both the consumer kind and
the
+ identifier the server keys the stored offset on.
+ """
+ @typing.final
+ class Single(Consumer):
+ r"""
+ A regular consumer, owning its offset on the polled partition.
+ """
+
+ __match_args__ = ("id",)
+ @property
+ def id(self) -> builtins.str | builtins.int: ...
+ def __new__(cls, id: builtins.str | builtins.int) -> Consumer.Single:
...
+
+ @typing.final
+ class Group(Consumer):
+ r"""
+ A member of the consumer group, sharing the group's offset.
+ """
+
+ __match_args__ = ("id",)
+ @property
+ def id(self) -> builtins.str | builtins.int: ...
+ def __new__(cls, id: builtins.str | builtins.int) -> Consumer.Group:
...
+
+ ...
+
@typing.final
class ConsumerGroup:
@property
@@ -358,6 +388,105 @@ class ConsumerGroupMember:
Gets the collection of partitions the consumer group member is
consuming.
"""
[email protected]
+class GlobalPermissions:
+ r"""
+ Global permissions, applied to all streams without specifying them one by
one.
+ """
+ @property
+ def manage_servers(self) -> builtins.bool:
+ r"""
+ Whether managing servers is allowed; includes `read_servers`.
+ """
+ @property
+ def read_servers(self) -> builtins.bool:
+ r"""
+ Whether reading server info (stats, clients) is allowed.
+ """
+ @property
+ def manage_users(self) -> builtins.bool:
+ r"""
+ Whether managing users is allowed; includes `read_users`.
+ """
+ @property
+ def read_users(self) -> builtins.bool:
+ r"""
+ Whether reading user info is allowed.
+ """
+ @property
+ def manage_streams(self) -> builtins.bool:
+ r"""
+ Whether managing all streams is allowed; includes `read_streams` and
+ `manage_topics`.
+ """
+ @property
+ def read_streams(self) -> builtins.bool:
+ r"""
+ Whether reading all streams is allowed; includes `read_topics`.
+ """
+ @property
+ def manage_topics(self) -> builtins.bool:
+ r"""
+ Whether managing all topics is allowed; includes `read_topics` and
+ `send_messages`.
+ """
+ @property
+ def read_topics(self) -> builtins.bool:
+ r"""
+ Whether reading all topics and managing consumer groups is allowed;
+ includes `poll_messages`.
+ """
+ @property
+ def poll_messages(self) -> builtins.bool:
+ r"""
+ Whether polling messages from all streams and managing consumer
+ offsets is allowed.
+ """
+ @property
+ def send_messages(self) -> builtins.bool:
+ r"""
+ Whether sending messages to all streams is allowed.
+ """
+ def __eq__(self, other: builtins.object, /) -> builtins.bool: ...
+ def __new__(
+ cls,
+ *,
+ manage_servers: builtins.bool = False,
+ read_servers: builtins.bool = False,
+ manage_users: builtins.bool = False,
+ read_users: builtins.bool = False,
+ manage_streams: builtins.bool = False,
+ read_streams: builtins.bool = False,
+ manage_topics: builtins.bool = False,
+ read_topics: builtins.bool = False,
+ poll_messages: builtins.bool = False,
+ send_messages: builtins.bool = False,
+ ) -> GlobalPermissions:
+ r"""
+ Create global permissions. Every flag defaults to `False`.
+
+ The `includes` notes below are transitive: a flag also grants
everything
+ its included flags grant. For example `manage_streams` includes
+ `manage_topics`, and through it `read_topics`, `poll_messages`, and
+ `send_messages`.
+
+ Args:
+ manage_servers: Allow managing servers; includes `read_servers`.
+ read_servers: Allow reading server info (stats, clients).
+ manage_users: Allow managing users; includes `read_users`.
+ read_users: Allow reading user info.
+ manage_streams: Allow managing all streams; includes `read_streams`
+ and `manage_topics`.
+ read_streams: Allow reading all streams; includes `read_topics`.
+ manage_topics: Allow managing all topics; includes `read_topics`
+ and `send_messages`.
+ read_topics: Allow reading all topics and managing consumer groups
+ (including create and delete); includes `poll_messages`.
+ poll_messages: Allow polling messages from all streams and managing
+ consumer offsets.
+ send_messages: Allow sending messages to all streams.
+ """
+
class HeaderKey:
r"""
Typed key for an Iggy user header.
@@ -708,105 +837,6 @@ class HeaderValue:
def value(self) -> builtins.float: ...
def __new__(cls, value: builtins.float) -> HeaderValue.Float64: ...
[email protected]
-class GlobalPermissions:
- r"""
- Global permissions, applied to all streams without specifying them one by
one.
- """
- @property
- def manage_servers(self) -> builtins.bool:
- r"""
- Whether managing servers is allowed; includes `read_servers`.
- """
- @property
- def read_servers(self) -> builtins.bool:
- r"""
- Whether reading server info (stats, clients) is allowed.
- """
- @property
- def manage_users(self) -> builtins.bool:
- r"""
- Whether managing users is allowed; includes `read_users`.
- """
- @property
- def read_users(self) -> builtins.bool:
- r"""
- Whether reading user info is allowed.
- """
- @property
- def manage_streams(self) -> builtins.bool:
- r"""
- Whether managing all streams is allowed; includes `read_streams` and
- `manage_topics`.
- """
- @property
- def read_streams(self) -> builtins.bool:
- r"""
- Whether reading all streams is allowed; includes `read_topics`.
- """
- @property
- def manage_topics(self) -> builtins.bool:
- r"""
- Whether managing all topics is allowed; includes `read_topics` and
- `send_messages`.
- """
- @property
- def read_topics(self) -> builtins.bool:
- r"""
- Whether reading all topics and managing consumer groups is allowed;
- includes `poll_messages`.
- """
- @property
- def poll_messages(self) -> builtins.bool:
- r"""
- Whether polling messages from all streams and managing consumer
- offsets is allowed.
- """
- @property
- def send_messages(self) -> builtins.bool:
- r"""
- Whether sending messages to all streams is allowed.
- """
- def __eq__(self, other: builtins.object, /) -> builtins.bool: ...
- def __new__(
- cls,
- *,
- manage_servers: builtins.bool = False,
- read_servers: builtins.bool = False,
- manage_users: builtins.bool = False,
- read_users: builtins.bool = False,
- manage_streams: builtins.bool = False,
- read_streams: builtins.bool = False,
- manage_topics: builtins.bool = False,
- read_topics: builtins.bool = False,
- poll_messages: builtins.bool = False,
- send_messages: builtins.bool = False,
- ) -> GlobalPermissions:
- r"""
- Create global permissions. Every flag defaults to `False`.
-
- The `includes` notes below are transitive: a flag also grants
everything
- its included flags grant. For example `manage_streams` includes
- `manage_topics`, and through it `read_topics`, `poll_messages`, and
- `send_messages`.
-
- Args:
- manage_servers: Allow managing servers; includes `read_servers`.
- read_servers: Allow reading server info (stats, clients).
- manage_users: Allow managing users; includes `read_users`.
- read_users: Allow reading user info.
- manage_streams: Allow managing all streams; includes `read_streams`
- and `manage_topics`.
- read_streams: Allow reading all streams; includes `read_topics`.
- manage_topics: Allow managing all topics; includes `read_topics`
- and `send_messages`.
- read_topics: Allow reading all topics and managing consumer groups
- (including create and delete); includes `poll_messages`.
- poll_messages: Allow polling messages from all streams and managing
- consumer offsets.
- send_messages: Allow sending messages to all streams.
- """
-
@typing.final
class IggyClient:
r"""
@@ -1315,13 +1345,17 @@ class IggyClient:
self,
stream: builtins.str | builtins.int,
topic: builtins.str | builtins.int,
- partition_id: builtins.int,
+ *,
+ consumer: Consumer,
polling_strategy: PollingStrategy,
count: builtins.int,
auto_commit: builtins.bool,
+ partition_id: builtins.int | None = None,
) -> collections.abc.Awaitable[list[ReceiveMessage]]:
r"""
- Polls for messages from the specified topic and partition.
+ Polls for messages from the specified topic on behalf of the given
consumer.
+ Omitting `partition_id` reads partition 0 for a regular consumer, and
+ polls the member's assigned partitions for a consumer group.
Returns a list of received messages or a RuntimeError on failure.
"""
def consumer_group(
@@ -1714,6 +1748,10 @@ class ReceiveMessage:
def user_headers(self) -> UserHeaders | None:
r"""
Retrieves user headers attached to the received message.
+
+ Returns `None` when no headers are present or when the headers
+ on the wire are structurally malformed (those errors are logged
+ internally). Only known semantic decode errors raise `ValueError`.
"""
@typing.final
@@ -2188,6 +2226,36 @@ class TopicPermissions:
send_messages: Allow sending messages to the topic.
"""
[email protected]
+class UserHeaders(dict):
+ r"""
+ User headers dictionary returned by `ReceiveMessage.user_headers`.
+
+ This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping
+ operations work) that additionally exposes `to_scalar_dict` for the
convenient
+ scalar form.
+ """
+ def __new__(cls, mapping: dict | None = None) -> UserHeaders:
+ r"""
+ Wraps a mapping so its entries gain the `to_scalar_dict` helper.
+
+ Accepts a dict whose keys and values can each independently be
+ `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool |
+ int | float`). The inherited `dict` initializer copies the provided
+ mapping.
+ """
+ def __setitem__(self, key: typing.Any, value: typing.Any) -> None: ...
+ def to_scalar_dict(
+ self,
+ ) -> dict[str | bytes | bool | int | float, str | bytes | bool | int |
float]:
+ r"""
+ Converts these headers into the convenient plain dictionary form.
+
+ Returns an error if two distinct typed keys map to the same plain
+ Python scalar (e.g., `UnsignedInt8(1)` and `UnsignedInt16(1)` both
+ become `int(1)`), or if a stored field cannot be decoded.
+ """
+
@typing.final
class UserInfo:
@property
@@ -2253,31 +2321,3 @@ class UserStatus(enum.Enum):
r"""
The user account is inactive and cannot be used.
"""
-
-class UserHeaders(dict):
- r"""
- User headers dictionary returned by `ReceiveMessage.user_headers`.
-
- This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping
- operations work) that additionally exposes `to_scalar_dict` for the
convenient
- scalar form.
- """
- def __new__(cls, mapping: dict | None = None) -> UserHeaders:
- r"""
- Wraps a mapping so its entries gain the `to_scalar_dict` helper.
-
- Accepts a dict whose keys and values can each independently be
- `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool |
- int | float`). The inherited `dict` initializer copies the provided
- mapping.
- """
- def to_scalar_dict(
- self,
- ) -> dict[str | bytes | bool | int | float, str | bytes | bool | int |
float]:
- r"""
- Converts these headers into the convenient plain dictionary form.
-
- Returns an error if two distinct typed keys map to the same plain
- Python scalar (e.g., `UnsignedInt8(1)` and `UnsignedInt16(1)` both
- become `int(1)`), or if a stored field cannot be decoded.
- """
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 816ba4e61..00f4122f3 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -33,8 +33,8 @@ use std::sync::Arc;
use crate::config::PyClientConfig;
use crate::consumer::{
- AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as
PyConsumerGroupDetails,
- IggyConsumer,
+ AutoCommit, Consumer as PyConsumer, ConsumerGroup as PyConsumerGroup,
+ ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer,
};
use crate::duration::{py_delta_to_iggy_duration, reject_zero};
use crate::identifier::PyIdentifier;
@@ -1035,21 +1035,25 @@ impl IggyClient {
})
}
- /// Polls for messages from the specified topic and partition.
+ /// Polls for messages from the specified topic on behalf of the given
consumer.
+ /// Omitting `partition_id` reads partition 0 for a regular consumer, and
+ /// polls the member's assigned partitions for a consumer group.
/// Returns a list of received messages or a RuntimeError on failure.
#[allow(clippy::too_many_arguments)]
+ #[pyo3(signature = (stream, topic, *, consumer, polling_strategy, count,
auto_commit, partition_id = None))]
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[ReceiveMessage]]",
imports=("collections.abc")))]
fn poll_messages<'a>(
&self,
py: Python<'a>,
stream: PyIdentifier,
topic: PyIdentifier,
- partition_id: u32,
+ consumer: &PyConsumer,
polling_strategy: &PollingStrategy,
count: u32,
auto_commit: bool,
+ partition_id: Option<u32>,
) -> PyResult<Bound<'a, PyAny>> {
- let consumer = RustConsumer::default();
+ let consumer = RustConsumer::try_from(consumer)?;
let stream = Identifier::try_from(stream)?;
let topic = Identifier::try_from(topic)?;
let strategy: RustPollingStrategy = polling_strategy.into();
@@ -1061,7 +1065,7 @@ impl IggyClient {
.poll_messages(
&stream,
&topic,
- Some(partition_id),
+ partition_id,
&consumer,
&strategy,
count,
@@ -1069,6 +1073,7 @@ impl IggyClient {
)
.await
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError,
_>(e.to_string()))?;
+ let partition_id = polled_messages.partition_id;
let messages = polled_messages
.messages
.into_iter()
diff --git a/foreign/python/src/consumer.rs b/foreign/python/src/consumer.rs
index 6a6e69a87..7f8fe8303 100644
--- a/foreign/python/src/consumer.rs
+++ b/foreign/python/src/consumer.rs
@@ -21,10 +21,10 @@ use futures::StreamExt;
use iggy::consumer_ext::{IggyConsumerMessageExt, MessageConsumer};
use iggy::prelude::{
AutoCommit as RustAutoCommit, AutoCommitAfter as RustAutoCommitAfter,
- AutoCommitWhen as RustAutoCommitWhen, ConsumerGroup as RustConsumerGroup,
- ConsumerGroupDetails as RustConsumerGroupDetails,
- ConsumerGroupMember as RustConsumerGroupMember, IggyConsumer as
RustIggyConsumer, IggyDuration,
- IggyError, ReceivedMessage,
+ AutoCommitWhen as RustAutoCommitWhen, Consumer as RustConsumer,
+ ConsumerGroup as RustConsumerGroup, ConsumerGroupDetails as
RustConsumerGroupDetails,
+ ConsumerGroupMember as RustConsumerGroupMember, Identifier, IggyConsumer
as RustIggyConsumer,
+ IggyDuration, IggyError, ReceivedMessage,
};
use pyo3::exceptions::PyStopAsyncIteration;
use pyo3::types::PyDelta;
@@ -215,6 +215,29 @@ impl IggyConsumer {
}
}
+/// The consumer polling the messages. It selects both the consumer kind and
the
+/// identifier the server keys the stored offset on.
+#[derive(Clone)]
+#[gen_stub_pyclass_complex_enum]
+#[pyclass(from_py_object)]
+pub enum Consumer {
+ /// A regular consumer, owning its offset on the polled partition.
+ Single { id: PyIdentifier },
+ /// A member of the consumer group, sharing the group's offset.
+ Group { id: PyIdentifier },
+}
+
+impl TryFrom<&Consumer> for RustConsumer {
+ type Error = PyErr;
+
+ fn try_from(value: &Consumer) -> PyResult<RustConsumer> {
+ Ok(match value {
+ Consumer::Single { id } =>
RustConsumer::new(Identifier::try_from(id)?),
+ Consumer::Group { id } =>
RustConsumer::group(Identifier::try_from(id)?),
+ })
+ }
+}
+
#[gen_stub_pyclass]
#[pyclass]
pub struct ConsumerGroup {
diff --git a/foreign/python/src/identifier.rs b/foreign/python/src/identifier.rs
index a113e5e54..f800e80bd 100644
--- a/foreign/python/src/identifier.rs
+++ b/foreign/python/src/identifier.rs
@@ -24,8 +24,8 @@ use pyo3::{
};
use pyo3_stub_gen::impl_stub_type;
-#[derive(FromPyObject, IntoPyObject)]
-pub(crate) enum PyIdentifier {
+#[derive(Clone, FromPyObject, IntoPyObject)]
+pub enum PyIdentifier {
#[pyo3(transparent, annotation = "str")]
String(String),
#[pyo3(transparent, annotation = "int")]
@@ -33,21 +33,29 @@ pub(crate) enum PyIdentifier {
}
impl_stub_type!(PyIdentifier = String | isize);
-impl TryFrom<PyIdentifier> for Identifier {
+impl TryFrom<&PyIdentifier> for Identifier {
type Error = PyErr;
- fn try_from(py_identifier: PyIdentifier) -> Result<Self, Self::Error> {
+ fn try_from(py_identifier: &PyIdentifier) -> Result<Self, Self::Error> {
match py_identifier {
PyIdentifier::String(s) => {
- Identifier::from_str(&s).map_err(|e|
PyErr::new::<PyValueError, _>(e.to_string()))
+ Identifier::from_str(s).map_err(|e| PyErr::new::<PyValueError,
_>(e.to_string()))
}
PyIdentifier::Int(i) => {
- Identifier::numeric(i).map_err(|e| PyErr::new::<PyValueError,
_>(e.to_string()))
+ Identifier::numeric(*i).map_err(|e| PyErr::new::<PyValueError,
_>(e.to_string()))
}
}
}
}
+impl TryFrom<PyIdentifier> for Identifier {
+ type Error = PyErr;
+
+ fn try_from(py_identifier: PyIdentifier) -> Result<Self, Self::Error> {
+ Identifier::try_from(&py_identifier)
+ }
+}
+
impl TryFrom<&Identifier> for PyIdentifier {
type Error = PyErr;
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 7910af77b..5f2e12826 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -32,7 +32,7 @@ mod user_headers;
use client::IggyClient;
use config::{AutoLogin, TcpConfig, TcpReconnectionConfig};
use consumer::{
- AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup,
ConsumerGroupDetails,
+ AutoCommit, AutoCommitAfter, AutoCommitWhen, Consumer, ConsumerGroup,
ConsumerGroupDetails,
ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator,
};
use options::OptionSpec;
@@ -63,6 +63,7 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) ->
PyResult<()> {
m.add_class::<MaxTopicSize>()?;
m.add_class::<OptionSpec>()?;
m.add_class::<Partition>()?;
+ m.add_class::<Consumer>()?;
m.add_class::<ConsumerGroup>()?;
m.add_class::<ConsumerGroupDetails>()?;
m.add_class::<ConsumerGroupMember>()?;
diff --git a/foreign/python/tests/test_message_operations.py
b/foreign/python/tests/test_message_operations.py
index d9128a0e1..bbf5b1d0e 100644
--- a/foreign/python/tests/test_message_operations.py
+++ b/foreign/python/tests/test_message_operations.py
@@ -21,6 +21,7 @@ import uuid
import pytest
from apache_iggy import (
+ Consumer,
HeaderKey,
HeaderValue,
IggyClient,
@@ -58,6 +59,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -93,6 +95,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -136,6 +139,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -170,6 +174,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -226,6 +231,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -276,6 +282,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -318,6 +325,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -619,6 +627,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -660,6 +669,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -699,6 +709,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -735,6 +746,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=1,
@@ -769,6 +781,7 @@ class TestMessageOperations:
polled_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -806,6 +819,7 @@ class TestMessageOperations:
await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=0,
@@ -843,6 +857,7 @@ class TestMessageOperations:
await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=1,
polling_strategy=PollingStrategy.First(),
count=1,
@@ -875,6 +890,7 @@ class TestMessageOperations:
last_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=1,
@@ -909,6 +925,7 @@ class TestMessageOperations:
last_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Last(),
count=2,
@@ -945,6 +962,7 @@ class TestMessageOperations:
first_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -955,6 +973,7 @@ class TestMessageOperations:
offset_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Offset(value=start_offset),
count=10,
@@ -991,6 +1010,7 @@ class TestMessageOperations:
first_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -1001,6 +1021,7 @@ class TestMessageOperations:
offset_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Offset(value=offset_beyond_newest),
count=10,
@@ -1036,6 +1057,7 @@ class TestMessageOperations:
first_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -1046,6 +1068,7 @@ class TestMessageOperations:
timestamp_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Timestamp(value=start_timestamp),
count=10,
@@ -1084,6 +1107,7 @@ class TestMessageOperations:
first_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -1094,6 +1118,7 @@ class TestMessageOperations:
timestamp_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Timestamp(value=timestamp_after_newest),
count=10,
@@ -1130,6 +1155,7 @@ class TestMessageOperations:
first_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -1150,6 +1176,7 @@ class TestMessageOperations:
next_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Next(),
count=10,
@@ -1188,6 +1215,7 @@ class TestMessageOperations:
first_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
@@ -1208,6 +1236,7 @@ class TestMessageOperations:
next_messages = await iggy_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.Next(),
count=10,
@@ -1217,3 +1246,351 @@ class TestMessageOperations:
assert [
message.payload().decode("utf-8") for message in next_messages
] == existing_messages + new_messages
+
+ @pytest.mark.asyncio
+ async def
test_poll_messages_with_distinct_consumers_keeps_offsets_independent(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test each consumer owns its offset, so both see the whole
partition."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ test_messages = [f"Consumer isolation {i} - {unique_name()}" for i in
range(3)]
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(message) for message in test_messages],
+ )
+
+ for consumer_name in ("isolation-consumer-a", "isolation-consumer-b"):
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Single(consumer_name),
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ assert [
+ message.payload().decode("utf-8") for message in
polled_messages
+ ] == test_messages
+
+ @pytest.mark.asyncio
+ async def test_poll_messages_with_shared_consumer_shares_the_partition(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test two polls under the same consumer share one stored offset."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ test_messages = [f"Shared consumer {i} - {unique_name()}" for i in
range(3)]
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(message) for message in test_messages],
+ )
+
+ first_poll = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Single("shared-consumer"),
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ assert [
+ message.payload().decode("utf-8") for message in first_poll
+ ] == test_messages
+
+ second_poll = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Single("shared-consumer"),
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ assert second_poll == []
+
+ @pytest.mark.asyncio
+ async def test_poll_messages_without_a_consumer_raises_type_error(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test the consumer argument is required."""
+ # Argument binding fails before the call reaches the server, so the
+ # stream and topic never have to exist.
+ with pytest.raises(TypeError, match="consumer"):
+ # pyrefly: ignore # missing-argument
+ await iggy_client.poll_messages(
+ stream=unique_name(),
+ topic=unique_name(),
+ partition_id=0,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "consumer", ["single", 1, None], ids=["str", "int", "none"]
+ )
+ async def test_poll_messages_with_a_non_consumer_raises_type_error(
+ self, iggy_client: IggyClient, unique_name, consumer
+ ):
+ """Test the consumer argument only accepts a Consumer."""
+ # Argument binding fails before the call reaches the server, so the
+ # stream and topic never have to exist.
+ with pytest.raises(TypeError, match="not an instance of 'Consumer'"):
+ # pyrefly: ignore # bad-argument-type
+ await iggy_client.poll_messages(
+ stream=unique_name(),
+ topic=unique_name(),
+ consumer=consumer,
+ partition_id=0,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+
+ @pytest.mark.parametrize("identifier", [-1, 2**32], ids=["negative",
"above-u32"])
+ def test_consumer_rejects_numeric_ids_outside_u32(self, identifier):
+ """Test an out-of-range numeric id is refused when the consumer is
built."""
+ with pytest.raises(TypeError):
+ Consumer.Single(identifier)
+ with pytest.raises(TypeError):
+ Consumer.Group(identifier)
+
+ def test_consumer_accepts_valid_numeric_ids(self):
+ """Test both consumer kinds keep the numeric id they were given."""
+ assert Consumer.Single(1).id == 1
+ assert Consumer.Group(4294967295).id == 4294967295
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("identifier", ["", "a" * 256], ids=["empty",
"too-long"])
+ @pytest.mark.parametrize(
+ "consumer_kind", [Consumer.Single, Consumer.Group], ids=["single",
"group"]
+ )
+ async def
test_poll_messages_with_an_invalid_string_identifier_raises_value_error(
+ self, iggy_client: IggyClient, unique_name, identifier, consumer_kind
+ ):
+ """Test a string id is validated when it is converted, not when it is
built."""
+ 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(ValueError, match="Invalid identifier"):
+ await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=consumer_kind(identifier),
+ partition_id=0,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+
+ @pytest.mark.asyncio
+ async def test_poll_messages_without_partition_id_reads_partition_zero(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test a regular consumer falls back to partition 0, not the whole
topic."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partitions_count = 3
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name,
partitions_count=partitions_count
+ )
+ for partition_id in range(partitions_count):
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(f"Partition {partition_id}")],
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Single("partition-zero-consumer"),
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+
+ assert [
+ (message.partition_id(), message.payload().decode("utf-8"))
+ for message in polled_messages
+ ] == [(0, "Partition 0")]
+
+ @pytest.mark.asyncio
+ async def
test_poll_messages_with_consumer_group_rotates_assigned_partitions(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test a sole group member reaches every partition over repeated
polls."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+ partitions_count = 3
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name,
partitions_count=partitions_count
+ )
+ await iggy_client.create_consumer_group(stream_name, topic_name,
group_name)
+ await iggy_client.join_consumer_group(stream_name, topic_name,
group_name)
+ for partition_id in range(partitions_count):
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(f"Partition {partition_id}")],
+ )
+
+ polled = []
+ for _ in range(partitions_count):
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Group(group_name),
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ polled.extend(
+ (message.partition_id(), message.payload().decode("utf-8"))
+ for message in polled_messages
+ )
+
+ assert polled == [
+ (partition_id, f"Partition {partition_id}")
+ for partition_id in range(partitions_count)
+ ]
+
+ @pytest.mark.asyncio
+ async def test_poll_messages_with_an_unjoined_consumer_group_fails(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test a group poll needs membership, since it has no assignment
without it."""
+ 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.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=0,
+ messages=[Message("Unjoined group")],
+ )
+
+ # The group exists; only the join is missing.
+ with pytest.raises(RuntimeError, match="was not found"):
+ await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Group(group_name),
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+
+ @pytest.mark.asyncio
+ async def
test_poll_messages_with_consumer_group_honours_an_explicit_partition(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test an explicit partition overrides the rotation for a group
member."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+ partitions_count = 3
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name,
partitions_count=partitions_count
+ )
+ await iggy_client.create_consumer_group(stream_name, topic_name,
group_name)
+ await iggy_client.join_consumer_group(stream_name, topic_name,
group_name)
+ for partition_id in range(partitions_count):
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(f"Partition {partition_id}")],
+ )
+
+ # Partition 0 is the fallback, so asking for the others is what proves
+ # the explicit id is honoured.
+ for requested_partition in (1, 2):
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Group(group_name),
+ partition_id=requested_partition,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ assert [
+ (message.partition_id(), message.payload().decode("utf-8"))
+ for message in polled_messages
+ ] == [(requested_partition, f"Partition {requested_partition}")]
+
+ @pytest.mark.asyncio
+ async def test_poll_messages_with_consumer_group_reads_assigned_partitions(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test a group member polls its assignment when no partition is
given."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ group_name = unique_name()
+ test_messages = [f"Group polling {i} - {unique_name()}" for i in
range(3)]
+
+ 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.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=0,
+ messages=[Message(message) for message in test_messages],
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Group(group_name),
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ assert [
+ message.payload().decode("utf-8") for message in polled_messages
+ ] == test_messages
diff --git a/foreign/python/tests/test_tls.py b/foreign/python/tests/test_tls.py
index f914e4146..f9be94379 100644
--- a/foreign/python/tests/test_tls.py
+++ b/foreign/python/tests/test_tls.py
@@ -40,7 +40,7 @@ import pytest
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs
-from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import Consumer, IggyClient, PollingStrategy
from apache_iggy import SendMessage as Message
from .utils import wait_for_ping, wait_for_server
@@ -143,6 +143,7 @@ class TestTlsConnectivity:
polled = await tls_client.poll_messages(
stream=stream_name,
topic=topic_name,
+ consumer=Consumer.Single(1),
partition_id=partition_id,
polling_strategy=PollingStrategy.First(),
count=10,
