You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/09/13 15:57:47 UTC
(iggy) branch master updated: feat(python): add high-level producer API (#4156)
This is an automated email from the ASF dual-hosted git repository.
spetz 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 2588e9926 feat(python): add high-level producer API (#4156)
2588e9926 is described below
commit 2588e99268e03905d47a040acca9fb8126a6cf52
Author: Gunther Xing <ji...@gmail.com>
AuthorDate: Sun Sep 13 23:57:42 2026 +0800
feat(python): add high-level producer API (#4156)
Closes #4019
---
examples/python/README.md | 38 +
examples/python/high-level/background_producer.py | 108 ++
examples/python/high-level/consumer.py | 93 ++
examples/python/high-level/producer.py | 122 ++
foreign/python/README.md | 149 ++
foreign/python/apache_iggy.pyi | 246 ++++
foreign/python/src/client.rs | 110 ++
foreign/python/src/lib.rs | 11 +
foreign/python/src/partitioning.rs | 27 +-
foreign/python/src/producer.rs | 1307 +++++++++++++++++
foreign/python/src/send_message.rs | 26 +-
foreign/python/tests/test_producer.py | 1608 +++++++++++++++++++++
12 files changed, 3830 insertions(+), 15 deletions(-)
diff --git a/examples/python/README.md b/examples/python/README.md
index 14ed2adb7..8bbc4f03d 100644
--- a/examples/python/README.md
+++ b/examples/python/README.md
@@ -56,6 +56,44 @@ pip install ../../foreign/python .
## Basic Examples
+### High-Level Producer and Consumer
+
+The Python high-level producer API is a port of the Rust high-level producer
+API. For detailed producer behavior and configuration, see the
+[Rust high-level SDK documentation](https://iggy.apache.org/docs/sdk/rust/high-level-sdk/).
+
+The high-level producer binds the destination once, initializes missing
+resources, applies producer-level batching and retry settings, and shuts down
+deterministically through an async context manager. The high-level consumer
+joins a consumer group, polls all assigned partitions, invokes an async handler,
+and commits each message after it has been handled.
+
+Run either producer first. Both create the stream and topic and publish 12
+messages for the consumer, which exits after receiving all of them. `producer.py`
+uses direct mode and waits for server confirmations; `background_producer.py`
+uses bounded background workers and flushes them on context-manager exit:
+
+```bash
+# Using uv
+uv run high-level/producer.py
+uv run high-level/consumer.py
+
+# Or use the background producer before starting the same consumer
+uv run high-level/background_producer.py
+uv run high-level/consumer.py
+
+# Without using uv
+python high-level/producer.py
+python high-level/consumer.py
+
+# Or use the background producer before starting the same consumer
+python high-level/background_producer.py
+python high-level/consumer.py
+```
+
+The existing examples below use the low-level `IggyClient.send_messages()` API
+and remain useful when each call needs to specify its own destination.
+
### Getting Started
Perfect introduction for newcomers to Iggy:
diff --git a/examples/python/high-level/background_producer.py b/examples/python/high-level/background_producer.py
new file mode 100644
index 000000000..41033ee32
--- /dev/null
+++ b/examples/python/high-level/background_producer.py
@@ -0,0 +1,108 @@
+# 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.
+
+import argparse
+import asyncio
+from datetime import timedelta
+from typing import NamedTuple
+
+from apache_iggy import (
+ BackgroundProducerConfig,
+ BackpressureMode,
+ IggyClient,
+ Partitioning,
+ ProducerSharding,
+ SendMessage,
+)
+from loguru import logger
+
+STREAM_NAME = "high-level-stream"
+TOPIC_NAME = "high-level-topic"
+MESSAGES_TO_SEND = 12
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy, for example "
+ "'iggy+tcp://iggy:iggy@127.0.0.1:8090'"
+ ),
+ default="iggy+tcp://iggy:iggy@127.0.0.1:8090",
+ nargs="?",
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+async def main() -> None:
+ args = parse_args()
+ client = IggyClient.from_connection_string(args.connection_string)
+
+ logger.info("Connecting to Iggy")
+ await client.connect()
+
+ producer = await client.producer(
+ STREAM_NAME,
+ TOPIC_NAME,
+ partitioning=Partitioning.balanced(),
+ mode=BackgroundProducerConfig(
+ num_shards=4,
+ linger_time=timedelta(milliseconds=10),
+ batch_size=1024 * 1024,
+ batch_length=100,
+ max_buffer_size=32 * 1024 * 1024,
+ failure_mode=BackpressureMode.block_with_timeout(timedelta(seconds=1)),
+ max_in_flight=4,
+ # Ordered keeps one destination on one sequential worker. Balanced
+ # uses all workers but does not guarantee order for one destination.
+ sharding=ProducerSharding.ORDERED,
+ ),
+ create_stream_if_not_exists=True,
+ create_topic_if_not_exists=True,
+ topic_partitions_count=3,
+ send_retries=3,
+ send_retry_interval=timedelta(seconds=1),
+ )
+
+ async with producer:
+ first = await producer.send_one(SendMessage("background message 0"))
+ logger.info(
+ "Dispatcher accepted the first message with {} confirmations",
+ len(first.confirmations),
+ )
+
+ messages = [
+ SendMessage(f"background message {index}")
+ for index in range(1, MESSAGES_TO_SEND)
+ ]
+ accepted = await producer.send(messages)
+ logger.info(
+ "Dispatcher accepted {} more messages with {} confirmations",
+ len(messages),
+ len(accepted.confirmations),
+ )
+
+ logger.info("Shutdown flushed every accepted message")
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/high-level/consumer.py b/examples/python/high-level/consumer.py
new file mode 100644
index 000000000..67253e2b3
--- /dev/null
+++ b/examples/python/high-level/consumer.py
@@ -0,0 +1,93 @@
+# 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.
+
+import argparse
+import asyncio
+from datetime import timedelta
+from typing import NamedTuple
+
+from apache_iggy import (
+ AutoCommit,
+ AutoCommitAfter,
+ IggyClient,
+ PollingStrategy,
+ ReceiveMessage,
+)
+from loguru import logger
+
+STREAM_NAME = "high-level-stream"
+TOPIC_NAME = "high-level-topic"
+CONSUMER_GROUP_NAME = "high-level-consumer"
+MESSAGES_TO_CONSUME = 12
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy, for example "
+ "'iggy+tcp://iggy:iggy@127.0.0.1:8090'"
+ ),
+ default="iggy+tcp://iggy:iggy@127.0.0.1:8090",
+ nargs="?",
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+async def main() -> None:
+ args = parse_args()
+ client = IggyClient.from_connection_string(args.connection_string)
+
+ logger.info("Connecting to Iggy")
+ await client.connect()
+
+ consumer = await client.consumer_group(
+ CONSUMER_GROUP_NAME,
+ STREAM_NAME,
+ TOPIC_NAME,
+ polling_strategy=PollingStrategy.First(),
+ batch_length=100,
+ auto_commit=AutoCommit.After(AutoCommitAfter.ConsumingEachMessage()),
+ poll_interval=timedelta(milliseconds=100),
+ )
+
+ shutdown_event = asyncio.Event()
+ consumed_messages = 0
+
+ async def handle_message(message: ReceiveMessage) -> None:
+ nonlocal consumed_messages
+ consumed_messages += 1
+ logger.info(
+ "Received message from partition {} at offset {}: {}",
+ message.partition_id(),
+ message.offset(),
+ message.payload().decode("utf-8"),
+ )
+ if consumed_messages == MESSAGES_TO_CONSUME:
+ shutdown_event.set()
+
+ await consumer.consume_messages(handle_message, shutdown_event)
+ logger.info("Consumed {} messages, exiting", consumed_messages)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/examples/python/high-level/producer.py b/examples/python/high-level/producer.py
new file mode 100644
index 000000000..ceb01da0a
--- /dev/null
+++ b/examples/python/high-level/producer.py
@@ -0,0 +1,122 @@
+# 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.
+
+import argparse
+import asyncio
+from datetime import timedelta
+from typing import NamedTuple
+
+from apache_iggy import DirectProducerConfig, IggyClient, Partitioning, SendMessage
+from loguru import logger
+
+STREAM_NAME = "high-level-stream"
+TOPIC_NAME = "high-level-topic"
+SEND_TO_STREAM_NAME = "high-level-send-to-stream"
+SEND_TO_TOPIC_NAME = "high-level-send-to-topic"
+
+
+class ArgNamespace(NamedTuple):
+ connection_string: str
+
+
+def parse_args() -> ArgNamespace:
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "connection_string",
+ help=(
+ "Connection string for Iggy, for example "
+ "'iggy+tcp://iggy:iggy@127.0.0.1:8090'"
+ ),
+ default="iggy+tcp://iggy:iggy@127.0.0.1:8090",
+ nargs="?",
+ )
+ return ArgNamespace(**vars(parser.parse_args()))
+
+
+async def ensure_send_to_destination(client: IggyClient) -> None:
+ if await client.get_stream(SEND_TO_STREAM_NAME) is None:
+ await client.create_stream(SEND_TO_STREAM_NAME)
+ if await client.get_topic(SEND_TO_STREAM_NAME, SEND_TO_TOPIC_NAME) is None:
+ await client.create_topic(
+ stream=SEND_TO_STREAM_NAME,
+ name=SEND_TO_TOPIC_NAME,
+ partitions_count=2,
+ )
+
+
+async def main() -> None:
+ args = parse_args()
+ client = IggyClient.from_connection_string(args.connection_string)
+
+ logger.info("Connecting to Iggy")
+ await client.connect()
+ await ensure_send_to_destination(client)
+
+ producer = await client.producer(
+ STREAM_NAME,
+ TOPIC_NAME,
+ partitioning=Partitioning.balanced(),
+ mode=DirectProducerConfig(
+ batch_length=100,
+ linger_time=timedelta(milliseconds=5),
+ ),
+ create_stream_if_not_exists=True,
+ create_topic_if_not_exists=True,
+ topic_partitions_count=3,
+ topic_message_expiry=None,
+ topic_max_size=None,
+ send_retries=3,
+ send_retry_interval=timedelta(seconds=1),
+ )
+
+ async with producer:
+ confirmation = await producer.send_one(SendMessage("single message"))
+ logger.info(
+ "Sent one message with {} confirmation(s)",
+ len(confirmation.confirmations),
+ )
+
+ messages = [SendMessage(f"batch message {index}") for index in range(10)]
+ confirmation = await producer.send(messages)
+ logger.info(
+ "Sent a batch with {} confirmation(s)",
+ len(confirmation.confirmations),
+ )
+
+ confirmation = await producer.send_with_partitioning(
+ [SendMessage("partitioned message")],
+ Partitioning.partition_id(1),
+ )
+ logger.info(
+ "Sent to a selected partition with {} confirmation(s)",
+ len(confirmation.confirmations),
+ )
+
+ confirmation = await producer.send_to(
+ SEND_TO_STREAM_NAME,
+ SEND_TO_TOPIC_NAME,
+ [SendMessage("message for another topic")],
+ Partitioning.partition_id(0),
+ )
+ logger.info(
+ "Sent to another topic with {} confirmation(s)",
+ len(confirmation.confirmations),
+ )
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/foreign/python/README.md b/foreign/python/README.md
index 1718d2d89..c727cb5ae 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md
@@ -209,6 +209,155 @@ async def main():
asyncio.run(main())
```
+## High-Level Producer
+
+The Python high-level producer API is a port of the Rust high-level producer
+API. For detailed producer semantics and configuration guidance, see the
+[Rust high-level SDK documentation](https://iggy.apache.org/docs/sdk/rust/high-level-sdk/).
+
+Use `IggyClient.producer()` when an application repeatedly publishes to one
+stream and topic. Producer creation is asynchronous because it initializes the
+destination before returning. By default, it creates a missing stream and topic,
+uses balanced partitioning, sends directly in batches of up to 1,000 messages,
+and retries failed sends up to three times with a one-second retry interval.
+
+The default mode is direct. Pass `BackgroundProducerConfig` to queue sends on
+background workers instead.
+
+```python
+import asyncio
+from datetime import timedelta
+
+from apache_iggy import DirectProducerConfig, IggyClient, Partitioning, SendMessage
+
+
+async def main():
+ client = IggyClient.from_connection_string("iggy+tcp://iggy:iggy@127.0.0.1:8090")
+ await client.connect()
+
+ producer = await client.producer(
+ "orders",
+ "created",
+ partitioning=Partitioning.balanced(),
+ mode=DirectProducerConfig(
+ batch_length=500,
+ linger_time=timedelta(milliseconds=5),
+ ),
+ create_stream_if_not_exists=True,
+ create_topic_if_not_exists=True,
+ topic_partitions_count=3,
+ topic_message_expiry=None,
+ topic_max_size=None,
+ send_retries=3,
+ send_retry_interval=timedelta(seconds=1),
+ )
+
+ async with producer:
+ await producer.send_one(SendMessage("order-1"))
+ response = await producer.send([SendMessage("order-2"), SendMessage("order-3")])
+ print(f"Received {len(response.confirmations)} partition confirmations")
+
+
+asyncio.run(main())
+```
+
+The producer is bound to the stream and topic passed to `producer()`. Use
+`send_with_partitioning(messages, partitioning)` to override its partitioning
+strategy for one call, or `send_to(stream, topic, messages, partitioning)` to
+send to another existing destination. `send_to()` does not create or initialize
+that destination.
+
+Direct sends use at-least-once delivery. A request can commit even when its
+response is lost, so any retry can write the same batch again. `send_retries`
+counts retries after the initial attempt. The first retry runs immediately, and
+`send_retry_interval` delays only later retries. Set `send_retries` to `None` or
+`0` to disable producer retries. Set `send_retry_interval` to `None` to run all
+enabled retries without a delay. A zero interval raises `ValueError`.
+
+Transport retries are separate from producer retries. For example, the default
+`HttpConfig(retries=3)` gives each producer attempt up to four HTTP attempts.
+
+A failed direct send raises `ProducerSendError`, which is a `RuntimeError`
+subclass. Its `cause` property contains the underlying error text. Its
+`committed` property contains confirmations from completed chunks, and `failed`
+contains the remaining unconfirmed messages. If encryption is enabled, the
+failed messages contain encrypted payloads. Restore the original payloads
+before submitting them to the same producer again.
+
+### Background Mode
+
+A successful background send means the dispatcher accepted the messages. The
+worker writes them later, so all four send methods return a
+`SendMessagesResponse` with an empty `confirmations` list. It does not mean the
+server has committed the messages.
+
+```python
+from datetime import timedelta
+
+from apache_iggy import (
+ BackgroundProducerConfig,
+ BackpressureMode,
+ ProducerSharding,
+ SendMessage,
+)
+
+producer = await client.producer(
+ "orders",
+ "created",
+ mode=BackgroundProducerConfig(
+ num_shards=4,
+ linger_time=timedelta(milliseconds=10),
+ batch_size=1024 * 1024,
+ batch_length=100,
+ max_buffer_size=32 * 1024 * 1024,
+ failure_mode=BackpressureMode.block_with_timeout(timedelta(seconds=1)),
+ max_in_flight=4,
+ sharding=ProducerSharding.ORDERED,
+ ),
+)
+
+async with producer:
+ accepted = await producer.send_one(SendMessage("order-1"))
+ assert accepted.confirmations == []
+```
+
+Each shard flushes when any configured condition is met: `batch_length` queued
+send calls, `batch_size` reported bytes, or `linger_time` since the first send
+entered an empty buffer. `batch_length` counts calls, not individual messages.
+Zero disables either batching threshold, while a zero linger flushes as soon as
+the worker receives a send.
+
+`ProducerSharding.ORDERED` hashes the stream/topic destination so its sends use
+one sequential worker and retain dispatch order. `ProducerSharding.BALANCED`
+assigns consecutive sends round-robin across the shards for throughput; ordering
+for one destination is not guaranteed.
+
+`max_buffer_size` bounds bytes queued or in flight across the whole producer.
+When it is full, `BackpressureMode.block()` waits indefinitely,
+`block_with_timeout(duration)` waits up to that duration, and
+`fail_immediately()` raises `RuntimeError` without accepting the batch. A single
+batch larger than the whole budget always fails. A zero byte budget is
+unlimited. `max_in_flight` separately bounds concurrent write requests across
+all shards; zero uses the runtime maximum.
+
+Producer retries and transport reconnection happen inside background workers.
+The Python API does not currently expose a background error callback, so a write
+that still fails after its retries is logged by the Rust SDK and its unconfirmed
+messages are dropped.
+
+See the complete runnable
+[`background_producer.py`](../../examples/python/high-level/background_producer.py)
+example for all background configuration fields and deterministic shutdown.
+
+Cleanup is asynchronous and must be explicit. Prefer `async with`, as above, so
+shutdown runs on both successful and exceptional exits. Otherwise, call
+`await producer.shutdown()` in a `finally` block. Shutdown waits for active
+sends, is safe to call more than once, and rejects later sends with
+`RuntimeError`. In background mode it also drains every queue and flushes all
+accepted messages before returning. Object destruction does not perform
+asynchronous cleanup; dropping a background producer without `shutdown()` can
+lose buffered messages.
+
## Examples
Refer to the [examples/python/](https://github.com/apache/iggy/tree/master/examples/python) directory for usage examples.
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index 1150fa018..e628a7c4d 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -31,12 +31,15 @@ __all__ = [
"AutoCommitAfter",
"AutoCommitWhen",
"AutoLogin",
+ "BackgroundProducerConfig",
+ "BackpressureMode",
"CacheMetrics",
"CacheMetricsKey",
"Consumer",
"ConsumerGroup",
"ConsumerGroupDetails",
"ConsumerGroupMember",
+ "DirectProducerConfig",
"GlobalPermissions",
"HeaderKey",
"HeaderValue",
@@ -44,12 +47,15 @@ __all__ = [
"IggyClient",
"IggyConsumer",
"IggyExpiry",
+ "IggyProducer",
"MaxTopicSize",
"OptionSpec",
"Partition",
"Partitioning",
"Permissions",
"PollingStrategy",
+ "ProducerSendError",
+ "ProducerSharding",
"QuicConfig",
"QuicReconnectionConfig",
"ReceiveMessage",
@@ -302,6 +308,107 @@ class AutoLogin:
"""
def __repr__(self) -> builtins.str: ...
+@typing.final
+class BackgroundProducerConfig:
+ r"""
+ Immutable configuration for a producer that queues sends on background workers.
+
+ For detailed background-producer semantics, see
+ https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
+ """
+ @property
+ def num_shards(self) -> builtins.int:
+ r"""
+ Number of background worker shards, each with its own queue.
+ A value of zero is treated as one shard.
+ """
+ @property
+ def linger_time(self) -> datetime.timedelta:
+ r"""
+ Maximum time a worker holds a non-empty buffer before flushing it.
+ A zero duration flushes as soon as the worker receives a send.
+ """
+ @property
+ def batch_size(self) -> builtins.int:
+ r"""
+ Per-worker flush threshold in buffered bytes.
+ A value of zero disables this threshold.
+ """
+ @property
+ def batch_length(self) -> builtins.int:
+ r"""
+ Per-worker flush threshold in queued sends, not individual messages.
+ A value of zero disables this threshold.
+ """
+ @property
+ def max_buffer_size(self) -> builtins.int:
+ r"""
+ Maximum bytes buffered or in flight across all worker shards.
+ A value of zero makes the byte budget unlimited.
+ """
+ @property
+ def failure_mode(self) -> BackpressureMode:
+ r"""
+ Behavior when `max_buffer_size` is exhausted.
+ """
+ @property
+ def max_in_flight(self) -> builtins.int:
+ r"""
+ Maximum number of requests written concurrently across all workers.
+ A value of zero uses the runtime's maximum semaphore permit count.
+ """
+ @property
+ def sharding(self) -> ProducerSharding:
+ r"""
+ Strategy used to assign each send to a worker shard.
+ Ordered sharding preserves per-destination dispatch order, while balanced
+ sharding distributes sends round-robin and may reorder them.
+ """
+ def __new__(
+ cls,
+ *,
+ num_shards: builtins.int = 1,
+ linger_time: datetime.timedelta = ...,
+ batch_size: builtins.int = 1048576,
+ batch_length: builtins.int = 1000,
+ max_buffer_size: builtins.int = 33554432,
+ failure_mode: BackpressureMode = ...,
+ max_in_flight: builtins.int = 1,
+ sharding: ProducerSharding = ...,
+ ) -> BackgroundProducerConfig:
+ r"""
+ Constructs background batching, capacity, backpressure, and sharding configuration.
+ """
+ def __repr__(self) -> builtins.str: ...
+
+@typing.final
+class BackpressureMode:
+ r"""
+ What a background send does when the producer buffer is full.
+ """
+ @property
+ def timeout(self) -> datetime.timedelta | None:
+ r"""
+ The configured timeout, or `None` for modes without one.
+ """
+ def __eq__(self, other: builtins.object, /) -> builtins.bool: ...
+ @staticmethod
+ def block() -> BackpressureMode:
+ r"""
+ Wait indefinitely for buffer capacity.
+ """
+ @staticmethod
+ def block_with_timeout(timeout: datetime.timedelta) -> BackpressureMode:
+ r"""
+ Wait up to `timeout` for buffer capacity.
+ """
+ @staticmethod
+ def fail_immediately() -> BackpressureMode:
+ r"""
+ Fail immediately when the producer buffer is full.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class CacheMetrics:
r"""
@@ -451,6 +558,30 @@ class ConsumerGroupMember:
Gets the collection of partitions the consumer group member is consuming.
"""
+@typing.final
+class DirectProducerConfig:
+ r"""
+ Configuration for a producer that sends from the calling task.
+ """
+ @property
+ def batch_length(self) -> builtins.int:
+ r"""
+ Maximum number of messages sent in one request.
+ A value of zero uses the internal limit of 1,000,000 messages.
+ """
+ @property
+ def linger_time(self) -> datetime.timedelta:
+ r"""
+ Minimum gap requested between sequential direct sends.
+ """
+ def __new__(
+ cls, *, batch_length: builtins.int = 1000, linger_time: datetime.timedelta = ...
+ ) -> DirectProducerConfig:
+ r"""
+ Constructs direct-producer batching and pacing configuration.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class GlobalPermissions:
r"""
@@ -1670,6 +1801,32 @@ class IggyClient:
the supported unsigned 32-bit range.
RuntimeError: If the request fails.
"""
+ def producer(
+ self,
+ stream: builtins.str,
+ topic: builtins.str,
+ partitioning: Partitioning | None = None,
+ mode: DirectProducerConfig | BackgroundProducerConfig | None = None,
+ create_stream_if_not_exists: builtins.bool = True,
+ create_topic_if_not_exists: builtins.bool = True,
+ topic_partitions_count: builtins.int = 1,
+ topic_message_expiry: IggyExpiry | None = None,
+ topic_max_size: MaxTopicSize | None = None,
+ send_retries: builtins.int | None = 3,
+ send_retry_interval: datetime.timedelta | None = ...,
+ ) -> collections.abc.Awaitable[IggyProducer]:
+ r"""
+ Creates and initializes a high-level producer bound to a stream and topic.
+
+ This is a Python port of the Rust high-level producer API. For detailed
+ producer semantics, see https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
+ `None` selects direct mode. `BackgroundProducerConfig` starts background
+ workers and makes successful sends mean queue acceptance rather than a
+ server commit. The returned producer is ready to send.
+
+ Raises `ValueError` for invalid names or numeric ranges and `RuntimeError`
+ when stream/topic initialization fails.
+ """
def poll_messages(
self,
stream: builtins.str | builtins.int,
@@ -1863,6 +2020,63 @@ class IggyExpiry:
...
+@typing.final
+class IggyProducer:
+ r"""
+ Python port of the Rust high-level producer API, bound to one stream and topic.
+
+ For detailed producer semantics, see
+ https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
+
+ Direct sends complete after the server responds and contain commit confirmations.
+ Background sends complete once accepted by a worker and contain no confirmations.
+ Always use the async context manager or call `shutdown()` explicitly; dropping a
+ background producer can lose accepted buffered messages.
+ """
+ def send(
+ self, messages: list[SendMessage]
+ ) -> collections.abc.Awaitable[SendMessagesResponse]:
+ r"""
+ Sends a batch to the producer's bound stream and topic.
+ In background mode, success means accepted into the dispatcher and the
+ returned confirmation list is empty.
+ """
+ def send_one(
+ self, message: SendMessage
+ ) -> collections.abc.Awaitable[SendMessagesResponse]:
+ r"""
+ Sends one message to the producer's bound stream and topic.
+ It has the same mode-dependent completion semantics as `send()`.
+ """
+ def send_with_partitioning(
+ self, messages: list[SendMessage], partitioning: Partitioning | None = None
+ ) -> collections.abc.Awaitable[SendMessagesResponse]:
+ r"""
+ Sends a batch with an optional per-call partitioning override.
+ It has the same mode-dependent completion semantics as `send()`.
+ """
+ def send_to(
+ self,
+ stream: builtins.str | builtins.int,
+ topic: builtins.str | builtins.int,
+ messages: list[SendMessage],
+ partitioning: Partitioning | None = None,
+ ) -> collections.abc.Awaitable[SendMessagesResponse]:
+ r"""
+ Sends a batch to another existing stream and topic.
+ It has the same mode-dependent completion semantics as `send()` and does
+ not create the alternate destination.
+ """
+ def shutdown(self) -> collections.abc.Awaitable[None]:
+ r"""
+ Waits for active sends and closes the producer. Repeated calls are safe.
+ Background shutdown flushes every accepted buffered message before returning.
+ """
+ def __aenter__(self) -> collections.abc.Awaitable[IggyProducer]: ...
+ def __aexit__(
+ self, _exc_type: typing.Any, _exc_value: typing.Any, _traceback: typing.Any
+ ) -> collections.abc.Awaitable[builtins.bool]: ...
+
class MaxTopicSize:
r"""
The maximum size of a topic.
@@ -2080,6 +2294,29 @@ class PollingStrategy:
...
+@typing.final
+class ProducerSendError(builtins.RuntimeError):
+ r"""
+ A direct producer error that preserves partial-send recovery state.
+ """
+ @property
+ def cause(self) -> builtins.str:
+ r"""
+ The underlying Iggy error message.
+ """
+ @property
+ def failed(self) -> builtins.list[SendMessage]:
+ r"""
+ Messages without a usable confirmation after the failure.
+ An encryptor can leave these messages encrypted, so do not submit them
+ to the same producer without restoring their original payloads.
+ """
+ @property
+ def committed(self) -> builtins.list[SendMessagesConfirmation]:
+ r"""
+ Confirmations returned for chunks committed before the failure.
+ """
+
@typing.final
class QuicConfig:
r"""
@@ -3248,6 +3485,15 @@ class WebSocketReconnectionConfig:
"""
def __repr__(self) -> builtins.str: ...
+@typing.final
+class ProducerSharding(enum.Enum):
+ r"""
+ How a background producer distributes sends among its workers.
+ """
+
+ ORDERED = ...
+ BALANCED = ...
+
@typing.final
class UserStatus(enum.Enum):
r"""
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 40e53bb0c..d75f7c350 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -42,6 +42,7 @@ use crate::identifier::PyIdentifier;
use crate::options::OptionSpec as PyOptionSpec;
use crate::partitioning::PyPartitioning;
use crate::permissions::Permissions as PyPermissions;
+use crate::producer::{IggyProducer, ProducerMode, RetryInterval, u32_param as producer_u32_param};
use crate::receive_message::{PollingStrategy, ReceiveMessage};
use crate::send_message::{SendMessage, SendMessagesResponse as PySendMessagesResponse};
use crate::stats::Stats as PyStats;
@@ -65,6 +66,10 @@ fn to_runtime_error<E: Display>(error: E) -> PyErr {
PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(error.to_string())
}
+fn to_value_error<E: Display>(error: E) -> PyErr {
+ PyErr::new::<pyo3::exceptions::PyValueError, _>(error.to_string())
+}
+
/// Resolves the shared `create_topic`/`update_topic` parameters, applying
/// server defaults where the caller left them unset.
fn resolve_topic_params(
@@ -1321,6 +1326,111 @@ impl IggyClient {
})
}
+ /// Creates and initializes a high-level producer bound to a stream and topic.
+ ///
+ /// This is a Python port of the Rust high-level producer API. For detailed
+ /// producer semantics, see https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
+ /// `None` selects direct mode. `BackgroundProducerConfig` starts background
+ /// workers and makes successful sends mean queue acceptance rather than a
+ /// server commit. The returned producer is ready to send.
+ ///
+ /// Raises `ValueError` for invalid names or numeric ranges and `RuntimeError`
+ /// when stream/topic initialization fails.
+ #[allow(clippy::too_many_arguments)]
+ #[pyo3(signature = (
+ stream,
+ topic,
+ partitioning=None,
+ mode=None,
+ create_stream_if_not_exists=true,
+ create_topic_if_not_exists=true,
+ topic_partitions_count=1,
+ topic_message_expiry=None,
+ topic_max_size=None,
+ send_retries=Some(3),
+ send_retry_interval=RetryInterval::default(),
+ ))]
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[IggyProducer]", imports=("collections.abc")))]
+ fn producer<'a>(
+ &self,
+ py: Python<'a>,
+ stream: &str,
+ topic: &str,
+ #[gen_stub(override_type(type_repr = "Partitioning | None"))] partitioning: Option<
+ &crate::partitioning::Partitioning,
+ >,
+ #[gen_stub(override_type(
+ type_repr = "DirectProducerConfig | BackgroundProducerConfig | None"
+ ))]
+ mode: Option<ProducerMode>,
+ create_stream_if_not_exists: bool,
+ create_topic_if_not_exists: bool,
+ topic_partitions_count: i64,
+ #[gen_stub(override_type(type_repr = "IggyExpiry | None"))] topic_message_expiry: Option<
+ &IggyExpiry,
+ >,
+ #[gen_stub(override_type(type_repr = "MaxTopicSize | None"))] topic_max_size: Option<
+ &MaxTopicSize,
+ >,
+ send_retries: Option<i64>,
+ send_retry_interval: RetryInterval,
+ ) -> PyResult<Bound<'a, PyAny>> {
+ let mode = mode.unwrap_or_default();
+
+ let topic_partitions_count =
+ producer_u32_param(topic_partitions_count, "topic_partitions_count")?;
+ let topic_message_expiry = topic_message_expiry
+ .map(RustIggyExpiry::try_from)
+ .transpose()?
+ .unwrap_or(RustIggyExpiry::ServerDefault);
+ let topic_max_size = topic_max_size
+ .map(RustMaxTopicSize::try_from)
+ .transpose()?
+ .unwrap_or(RustMaxTopicSize::ServerDefault);
+ let send_retries = send_retries
+ .map(|retries| producer_u32_param(retries, "send_retries"))
+ .transpose()?
+ .filter(|retries| *retries != 0);
+ let send_retry_interval = send_retry_interval.resolve()?;
+
+ let mut builder = self
+ .inner
+ .producer(stream, topic)
+ .map_err(to_value_error)?
+ .send_retries(send_retries, send_retry_interval);
+
+ builder = match mode {
+ ProducerMode::Direct(config) => builder.direct((&config).into()),
+ ProducerMode::Background(config) => builder.background((&config).try_into()?),
+ };
+
+ if let Some(partitioning) = partitioning {
+ builder = builder.partitioning(partitioning.inner.as_ref().clone());
+ }
+ if create_stream_if_not_exists {
+ builder = builder.create_stream_if_not_exists();
+ } else {
+ builder = builder.do_not_create_stream_if_not_exists();
+ }
+ if create_topic_if_not_exists {
+ builder = builder.create_topic_if_not_exists(
+ topic_partitions_count,
+ topic_message_expiry,
+ topic_max_size,
+ );
+ } else {
+ builder = builder.do_not_create_topic_if_not_exists();
+ }
+
+ future_into_py(py, async move {
+ // A background build starts Tokio worker tasks, so it must happen
+ // while this future is executing on the Rust runtime.
+ let producer = builder.build();
+ producer.init().await.map_err(to_runtime_error)?;
+ Ok(IggyProducer::new(producer))
+ })
+ }
+
/// 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.
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 4dbcd48c0..6462b786d 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -24,6 +24,7 @@ mod identifier;
mod options;
mod partitioning;
mod permissions;
+mod producer;
mod receive_message;
mod send_message;
mod stats;
@@ -44,6 +45,10 @@ use consumer::{
use options::OptionSpec;
use partitioning::Partitioning;
use permissions::{GlobalPermissions, Permissions, StreamPermissions, TopicPermissions};
+use producer::{
+ BackgroundProducerConfig, BackpressureMode, DirectProducerConfig, IggyProducer,
+ ProducerSendError, ProducerSharding,
+};
use pyo3::prelude::*;
use receive_message::{PollingStrategy, ReceiveMessage};
use send_message::{SendMessage, SendMessagesConfirmation, SendMessagesResponse};
@@ -62,6 +67,12 @@ fn apache_iggy(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<SendMessagesConfirmation>()?;
m.add_class::<ReceiveMessage>()?;
m.add_class::<IggyClient>()?;
+ m.add_class::<IggyProducer>()?;
+ m.add_class::<ProducerSendError>()?;
+ m.add_class::<DirectProducerConfig>()?;
+ m.add_class::<BackgroundProducerConfig>()?;
+ m.add_class::<ProducerSharding>()?;
+ m.add_class::<BackpressureMode>()?;
m.add_class::<AutoLogin>()?;
m.add_class::<TcpConfig>()?;
m.add_class::<TcpReconnectionConfig>()?;
diff --git a/foreign/python/src/partitioning.rs b/foreign/python/src/partitioning.rs
index 83d9715d8..bb345cf51 100644
--- a/foreign/python/src/partitioning.rs
+++ b/foreign/python/src/partitioning.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+use std::sync::Arc;
+
use iggy::prelude::Partitioning as RustPartitioning;
use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
use pyo3_stub_gen::{
@@ -27,7 +29,7 @@ use pyo3_stub_gen::{
#[pyclass(from_py_object)]
#[gen_stub_pyclass]
pub struct Partitioning {
- pub(crate) inner: RustPartitioning,
+ pub(crate) inner: Arc<RustPartitioning>,
}
#[gen_stub_pymethods]
@@ -37,7 +39,7 @@ impl Partitioning {
#[staticmethod]
pub fn balanced() -> Self {
Self {
- inner: RustPartitioning::balanced(),
+ inner: Arc::new(RustPartitioning::balanced()),
}
}
@@ -53,7 +55,7 @@ impl Partitioning {
#[staticmethod]
pub fn partition_id(partition_id: u32) -> Self {
Self {
- inner: RustPartitioning::partition_id(partition_id),
+ inner: Arc::new(RustPartitioning::partition_id(partition_id)),
}
}
@@ -73,7 +75,9 @@ impl Partitioning {
};
let inner = RustPartitioning::messages_key(&key)
.map_err(|error| PyValueError::new_err(error.to_string()))?;
- Ok(Self { inner })
+ Ok(Self {
+ inner: Arc::new(inner),
+ })
}
}
@@ -98,8 +102,21 @@ impl_stub_type!(PyPartitioning = Partitioning | isize);
impl From<PyPartitioning> for RustPartitioning {
fn from(partitioning: PyPartitioning) -> Self {
match partitioning {
- PyPartitioning::Strategy(partitioning) => partitioning.inner,
+ PyPartitioning::Strategy(partitioning) => partitioning.inner.as_ref().clone(),
PyPartitioning::PartitionId(partition_id) => Self::partition_id(partition_id),
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn clone_shares_the_rust_partitioning() {
+ let partitioning = Partitioning::balanced();
+ let cloned = partitioning.clone();
+
+ assert!(Arc::ptr_eq(&partitioning.inner, &cloned.inner));
+ }
+}
diff --git a/foreign/python/src/producer.rs b/foreign/python/src/producer.rs
new file mode 100644
index 000000000..c51e34d2f
--- /dev/null
+++ b/foreign/python/src/producer.rs
@@ -0,0 +1,1307 @@
+// 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, sync::Arc};
+
+use iggy::clients::producer_config::BackpressureMode as RustBackpressureMode;
+use iggy::prelude::{
+ BackgroundConfig as RustBackgroundConfig, BalancedSharding, DirectConfig as RustDirectConfig,
+ Identifier, IggyByteSize, IggyDuration, IggyError, IggyMessage as RustIggyMessage,
+ IggyProducer as RustIggyProducer, OrderedSharding,
+ SendMessagesConfirmationResponse as RustSendMessagesConfirmationResponse, Sharding,
+};
+use pyo3::IntoPyObjectExt;
+use pyo3::conversion::FromPyObject;
+use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
+use pyo3::prelude::*;
+use pyo3::types::{PyAny, PyDelta, PyInt, PyList, PyString};
+use pyo3_async_runtimes::tokio::future_into_py;
+use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pyclass_enum, gen_stub_pymethods};
+use pyo3_stub_gen::{PyStubType, TypeInfo};
+use tokio::sync::{RwLock, Semaphore};
+
+use crate::duration::{duration_repr, iggy_duration_to_py_delta, py_delta_to_iggy_duration};
+use crate::partitioning::Partitioning;
+use crate::send_message::{SendMessage, SendMessagesConfirmation, SendMessagesResponse};
+
+const DEFAULT_BACKGROUND_NUM_SHARDS: usize = 1;
+const DEFAULT_BACKGROUND_BATCH_SIZE: usize = 1024 * 1024;
+const DEFAULT_BACKGROUND_BATCH_LENGTH: usize = 1_000;
+const DEFAULT_BACKGROUND_MAX_BUFFER_SIZE: u64 = 32 * 1024 * 1024;
+const DEFAULT_BACKGROUND_MAX_IN_FLIGHT: usize = 1;
+
+/// Configuration for a producer that sends from the calling task.
+#[derive(Clone)]
+#[gen_stub_pyclass]
+#[pyclass(frozen, from_py_object)]
+pub struct DirectProducerConfig {
+ pub(crate) inner: RustDirectConfig,
+}
+
+impl Default for DirectProducerConfig {
+ fn default() -> Self {
+ Self {
+ inner: RustDirectConfig::builder().build(),
+ }
+ }
+}
+
+impl From<&DirectProducerConfig> for RustDirectConfig {
+ fn from(config: &DirectProducerConfig) -> Self {
+ config.inner.clone()
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl DirectProducerConfig {
+ /// Constructs direct-producer batching and pacing configuration.
+ #[new]
+ #[pyo3(signature = (*, batch_length=1000, linger_time=DefaultDuration::default()))]
+ fn new(batch_length: i64, linger_time: DefaultDuration) -> PyResult<Self> {
+ let batch_length = u32_param(batch_length, "batch_length")?;
+ let linger_time = linger_time.resolve(IggyDuration::from(0))?;
+ if linger_time.get_duration().as_micros() > u128::from(u64::MAX) {
+ return Err(PyValueError::new_err(format!(
+ "'linger_time' must not exceed {} microseconds",
+ u64::MAX
+ )));
+ }
+ Ok(Self {
+ inner: RustDirectConfig::builder()
+ .batch_length(batch_length)
+ .linger_time(linger_time)
+ .build(),
+ })
+ }
+
+ /// Maximum number of messages sent in one request.
+ /// A value of zero uses the internal limit of 1,000,000 messages.
+ #[getter]
+ fn batch_length(&self) -> u32 {
+ self.inner.batch_length
+ }
+
+ /// Minimum gap requested between sequential direct sends.
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn linger_time<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.linger_time)
+ }
+
+ fn __repr__(&self) -> String {
+ format!(
+ "DirectProducerConfig(batch_length={}, linger_time={})",
+ self.inner.batch_length,
+ duration_repr(self.inner.linger_time)
+ )
+ }
+}
+
+/// How a background producer distributes sends among its workers.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[gen_stub_pyclass_enum]
+#[pyclass(eq, from_py_object, rename_all = "UPPERCASE")]
+pub enum ProducerSharding {
+ Ordered,
+ Balanced,
+}
+
+impl From<ProducerSharding> for Box<dyn Sharding + Send + Sync> {
+ fn from(sharding: ProducerSharding) -> Self {
+ match sharding {
+ ProducerSharding::Ordered => Box::new(OrderedSharding),
+ ProducerSharding::Balanced => Box::new(BalancedSharding::default()),
+ }
+ }
+}
+
+/// What a background send does when the producer buffer is full.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[gen_stub_pyclass]
+#[pyclass(eq, frozen, from_py_object)]
+pub struct BackpressureMode {
+ kind: BackpressureKind,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum BackpressureKind {
+ Block,
+ BlockWithTimeout(IggyDuration),
+ FailImmediately,
+}
+
+impl From<&BackpressureMode> for RustBackpressureMode {
+ fn from(mode: &BackpressureMode) -> Self {
+ match mode.kind {
+ BackpressureKind::Block => Self::Block,
+ BackpressureKind::BlockWithTimeout(timeout) => Self::BlockWithTimeout(timeout),
+ BackpressureKind::FailImmediately => Self::FailImmediately,
+ }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl BackpressureMode {
+ /// Wait indefinitely for buffer capacity.
+ #[staticmethod]
+ fn block() -> Self {
+ Self {
+ kind: BackpressureKind::Block,
+ }
+ }
+
+ /// Wait up to `timeout` for buffer capacity.
+ #[staticmethod]
+ fn block_with_timeout(
+ #[gen_stub(override_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ timeout: Py<PyDelta>,
+ ) -> PyResult<Self> {
+ Ok(Self {
+ kind: BackpressureKind::BlockWithTimeout(py_delta_to_iggy_duration(&timeout)?),
+ })
+ }
+
+ /// Fail immediately when the producer buffer is full.
+ #[staticmethod]
+ fn fail_immediately() -> Self {
+ Self {
+ kind: BackpressureKind::FailImmediately,
+ }
+ }
+
+ /// The configured timeout, or `None` for modes without one.
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta | None", imports=("datetime")))]
+ #[getter]
+ fn timeout<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyDelta>>> {
+ match self.kind {
+ BackpressureKind::BlockWithTimeout(timeout) => {
+ iggy_duration_to_py_delta(py, timeout).map(Some)
+ }
+ BackpressureKind::Block | BackpressureKind::FailImmediately => Ok(None),
+ }
+ }
+
+ fn __repr__(&self) -> String {
+ match self.kind {
+ BackpressureKind::Block => "BackpressureMode.block()".to_owned(),
+ BackpressureKind::BlockWithTimeout(timeout) => format!(
+ "BackpressureMode.block_with_timeout({})",
+ duration_repr(timeout)
+ ),
+ BackpressureKind::FailImmediately => "BackpressureMode.fail_immediately()".to_owned(),
+ }
+ }
+}
+
+/// Immutable configuration for a producer that queues sends on background workers.
+///
+/// For detailed background-producer semantics, see
+/// https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
+#[derive(Clone)]
+#[gen_stub_pyclass]
+#[pyclass(frozen, from_py_object)]
+pub struct BackgroundProducerConfig {
+ num_shards: usize,
+ linger_time: IggyDuration,
+ batch_size: usize,
+ batch_length: usize,
+ max_buffer_size: IggyByteSize,
+ failure_mode: BackpressureMode,
+ max_in_flight: usize,
+ sharding: ProducerSharding,
+}
+
+impl Default for BackgroundProducerConfig {
+ fn default() -> Self {
+ Self {
+ num_shards: DEFAULT_BACKGROUND_NUM_SHARDS,
+ linger_time: IggyDuration::from(1_000),
+ batch_size: DEFAULT_BACKGROUND_BATCH_SIZE,
+ batch_length: DEFAULT_BACKGROUND_BATCH_LENGTH,
+ max_buffer_size: IggyByteSize::from(DEFAULT_BACKGROUND_MAX_BUFFER_SIZE),
+ failure_mode: BackpressureMode::block(),
+ max_in_flight: DEFAULT_BACKGROUND_MAX_IN_FLIGHT,
+ sharding: ProducerSharding::Ordered,
+ }
+ }
+}
+
+impl TryFrom<&BackgroundProducerConfig> for RustBackgroundConfig {
+ type Error = PyErr;
+
+ fn try_from(config: &BackgroundProducerConfig) -> PyResult<Self> {
+ let max_buffer_size = config.max_buffer_size.as_bytes_u64();
+ if max_buffer_size != 0 && max_buffer_size > Semaphore::MAX_PERMITS as u64 {
+ return Err(PyValueError::new_err(format!(
+ "'max_buffer_size' must not exceed {}",
+ Semaphore::MAX_PERMITS
+ )));
+ }
+ if config.max_in_flight != 0 && config.max_in_flight > Semaphore::MAX_PERMITS {
+ return Err(PyValueError::new_err(format!(
+ "'max_in_flight' must not exceed {}",
+ Semaphore::MAX_PERMITS
+ )));
+ }
+
+ Ok(RustBackgroundConfig::builder()
+ .num_shards(config.num_shards)
+ .linger_time(config.linger_time)
+ .batch_size(config.batch_size)
+ .batch_length(config.batch_length)
+ .max_buffer_size(config.max_buffer_size)
+ .failure_mode((&config.failure_mode).into())
+ .max_in_flight(config.max_in_flight)
+ .sharding(config.sharding.into())
+ .build())
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl BackgroundProducerConfig {
+ /// Constructs background batching, capacity, backpressure, and sharding configuration.
+ #[new]
+ #[allow(clippy::too_many_arguments)]
+ #[pyo3(signature = (
+ *,
+ num_shards=1,
+ linger_time=DefaultDuration::one_millisecond(),
+ batch_size=1048576,
+ batch_length=1000,
+ max_buffer_size=33554432,
+ failure_mode=DefaultBackpressureMode::default(),
+ max_in_flight=1,
+ sharding=DefaultProducerSharding::default(),
+ ))]
+ fn new(
+ num_shards: i128,
+ linger_time: DefaultDuration,
+ batch_size: i128,
+ batch_length: i128,
+ max_buffer_size: i128,
+ failure_mode: DefaultBackpressureMode,
+ max_in_flight: i128,
+ sharding: DefaultProducerSharding,
+ ) -> PyResult<Self> {
+ let linger_time = linger_time.resolve(IggyDuration::from(1_000))?;
+ Ok(Self {
+ num_shards: usize_param(num_shards, "num_shards")?,
+ linger_time,
+ batch_size: usize_param(batch_size, "batch_size")?,
+ batch_length: usize_param(batch_length, "batch_length")?,
+ max_buffer_size: IggyByteSize::from(u64_param(max_buffer_size, "max_buffer_size")?),
+ failure_mode: failure_mode.resolve(),
+ max_in_flight: usize_param(max_in_flight, "max_in_flight")?,
+ sharding: sharding.resolve(),
+ })
+ }
+
+ /// Number of background worker shards, each with its own queue.
+ /// A value of zero is treated as one shard.
+ #[getter]
+ fn num_shards(&self) -> usize {
+ self.num_shards
+ }
+
+ /// Maximum time a worker holds a non-empty buffer before flushing it.
+ /// A zero duration flushes as soon as the worker receives a send.
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))]
+ #[getter]
+ fn linger_time<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.linger_time)
+ }
+
+ /// Per-worker flush threshold in buffered bytes.
+ /// A value of zero disables this threshold.
+ #[getter]
+ fn batch_size(&self) -> usize {
+ self.batch_size
+ }
+
+ /// Per-worker flush threshold in queued sends, not individual messages.
+ /// A value of zero disables this threshold.
+ #[getter]
+ fn batch_length(&self) -> usize {
+ self.batch_length
+ }
+
+ /// Maximum bytes buffered or in flight across all worker shards.
+ /// A value of zero makes the byte budget unlimited.
+ #[getter]
+ fn max_buffer_size(&self) -> u64 {
+ self.max_buffer_size.as_bytes_u64()
+ }
+
+ /// Behavior when `max_buffer_size` is exhausted.
+ #[getter]
+ fn failure_mode(&self) -> BackpressureMode {
+ self.failure_mode.clone()
+ }
+
+ /// Maximum number of requests written concurrently across all workers.
+ /// A value of zero uses the runtime's maximum semaphore permit count.
+ #[getter]
+ fn max_in_flight(&self) -> usize {
+ self.max_in_flight
+ }
+
+ /// Strategy used to assign each send to a worker shard.
+ /// Ordered sharding preserves per-destination dispatch order, while balanced
+ /// sharding distributes sends round-robin and may reorder them.
+ #[getter]
+ fn sharding(&self) -> ProducerSharding {
+ self.sharding
+ }
+
+ fn __repr__(&self) -> String {
+ let sharding = match self.sharding {
+ ProducerSharding::Ordered => "ProducerSharding.ORDERED",
+ ProducerSharding::Balanced => "ProducerSharding.BALANCED",
+ };
+ format!(
+ "BackgroundProducerConfig(num_shards={}, linger_time={}, batch_size={}, batch_length={}, max_buffer_size={}, failure_mode={}, max_in_flight={}, sharding={sharding})",
+ self.num_shards,
+ duration_repr(self.linger_time),
+ self.batch_size,
+ self.batch_length,
+ self.max_buffer_size.as_bytes_u64(),
+ self.failure_mode.__repr__(),
+ self.max_in_flight,
+ )
+ }
+}
+
+/// A direct producer error that preserves partial-send recovery state.
+#[gen_stub_pyclass]
+#[pyclass(frozen, extends=PyRuntimeError)]
+pub struct ProducerSendError {
+ cause: String,
+ failed: Arc<Vec<RustIggyMessage>>,
+ committed: Arc<Vec<RustSendMessagesConfirmationResponse>>,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl ProducerSendError {
+ /// The underlying Iggy error message.
+ #[getter]
+ fn cause(&self) -> &str {
+ &self.cause
+ }
+
+ /// Messages without a usable confirmation after the failure.
+ /// An encryptor can leave these messages encrypted, so do not submit them
+ /// to the same producer without restoring their original payloads.
+ #[getter]
+ fn failed(&self) -> Vec<SendMessage> {
+ self.failed
+ .iter()
+ .map(SendMessage::clone_from_rust)
+ .collect()
+ }
+
+ /// Confirmations returned for chunks committed before the failure.
+ #[getter]
+ fn committed(&self) -> Vec<SendMessagesConfirmation> {
+ self.committed
+ .iter()
+ .map(SendMessagesConfirmation::from)
+ .collect()
+ }
+}
+
+impl ProducerSendError {
+ fn new_err(
+ cause: IggyError,
+ failed: Arc<Vec<RustIggyMessage>>,
+ committed: Arc<Vec<RustSendMessagesConfirmationResponse>>,
+ ) -> PyErr {
+ Python::attach(|py| {
+ let cause = cause.to_string();
+ let message = format!("Producer send failed: {cause}");
+ let cause_error = PyRuntimeError::new_err(cause.clone());
+ let instance = match Bound::new(
+ py,
+ Self {
+ cause,
+ failed,
+ committed,
+ },
+ ) {
+ Ok(instance) => instance,
+ Err(error) => return error,
+ };
+ if let Err(error) = instance.setattr("args", (message,)) {
+ return error;
+ }
+ let error = PyErr::from_value(instance.into_any());
+ error.set_cause(py, Some(cause_error));
+ error
+ })
+ }
+}
+
+/// Python port of the Rust high-level producer API, bound to one stream and topic.
+///
+/// For detailed producer semantics, see
+/// https://iggy.apache.org/docs/sdk/rust/high-level-sdk/.
+///
+/// Direct sends complete after the server responds and contain commit confirmations.
+/// Background sends complete once accepted by a worker and contain no confirmations.
+/// Always use the async context manager or call `shutdown()` explicitly; dropping a
+/// background producer can lose accepted buffered messages.
+#[derive(Clone)]
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+pub struct IggyProducer {
+ inner: Arc<RwLock<Option<RustIggyProducer>>>,
+}
+
+impl IggyProducer {
+ pub(crate) fn new(producer: RustIggyProducer) -> Self {
+ Self {
+ inner: Arc::new(RwLock::new(Some(producer))),
+ }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl IggyProducer {
+ /// Sends a batch to the producer's bound stream and topic.
+ /// In background mode, success means accepted into the dispatcher and the
+ /// returned confirmation list is empty.
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[SendMessagesResponse]", imports=("collections.abc")))]
+ fn send<'py>(
+ &self,
+ py: Python<'py>,
+ #[gen_stub(override_type(type_repr = "list[SendMessage]"))] messages: &Bound<'_, PyList>,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let messages = extract_messages(messages)?;
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ // Every send keeps a read guard for its full future. Concurrent sends
+ // remain possible while shutdown cannot consume an active producer.
+ let producer = inner.read().await;
+ let producer = producer
+ .as_ref()
+ .ok_or_else(|| PyRuntimeError::new_err("producer has been shut down"))?;
+ producer
+ .send(messages)
+ .await
+ .map(SendMessagesResponse::from)
+ .map_err(to_send_error)
+ })
+ }
+
+ /// Sends one message to the producer's bound stream and topic.
+ /// It has the same mode-dependent completion semantics as `send()`.
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[SendMessagesResponse]", imports=("collections.abc")))]
+ fn send_one<'py>(
+ &self,
+ py: Python<'py>,
+ message: PyRef<'_, SendMessage>,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let message = clone_rust_message(&message);
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ let producer = inner.read().await;
+ let producer = producer
+ .as_ref()
+ .ok_or_else(|| PyRuntimeError::new_err("producer has been shut down"))?;
+ producer
+ .send_one(message)
+ .await
+ .map(SendMessagesResponse::from)
+ .map_err(to_send_error)
+ })
+ }
+
+ /// Sends a batch with an optional per-call partitioning override.
+ /// It has the same mode-dependent completion semantics as `send()`.
+ #[pyo3(signature = (messages, partitioning=None))]
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[SendMessagesResponse]", imports=("collections.abc")))]
+ fn send_with_partitioning<'py>(
+ &self,
+ py: Python<'py>,
+ #[gen_stub(override_type(type_repr = "list[SendMessage]"))] messages: &Bound<'_, PyList>,
+ #[gen_stub(override_type(type_repr = "Partitioning | None"))] partitioning: Option<
+ &Partitioning,
+ >,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let messages = extract_messages(messages)?;
+ let partitioning = partitioning.map(|value| value.inner.clone());
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ let producer = inner.read().await;
+ let producer = producer
+ .as_ref()
+ .ok_or_else(|| PyRuntimeError::new_err("producer has been shut down"))?;
+ producer
+ .send_with_partitioning(messages, partitioning)
+ .await
+ .map(SendMessagesResponse::from)
+ .map_err(to_send_error)
+ })
+ }
+
+ /// Sends a batch to another existing stream and topic.
+ /// It has the same mode-dependent completion semantics as `send()` and does
+ /// not create the alternate destination.
+ #[pyo3(signature = (stream, topic, messages, partitioning=None))]
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[SendMessagesResponse]", imports=("collections.abc")))]
+ fn send_to<'py>(
+ &self,
+ py: Python<'py>,
+ #[gen_stub(override_type(type_repr = "builtins.str | builtins.int"))] stream: &Bound<
+ '_,
+ PyAny,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.str | builtins.int"))] topic: &Bound<
+ '_,
+ PyAny,
+ >,
+ #[gen_stub(override_type(type_repr = "list[SendMessage]"))] messages: &Bound<'_, PyList>,
+ #[gen_stub(override_type(type_repr = "Partitioning | None"))] partitioning: Option<
+ &Partitioning,
+ >,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let stream = Arc::new(extract_send_to_identifier(stream, "stream")?);
+ let topic = Arc::new(extract_send_to_identifier(topic, "topic")?);
+ let messages = extract_messages(messages)?;
+ let partitioning = partitioning.map(|value| value.inner.clone());
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ let producer = inner.read().await;
+ let producer = producer
+ .as_ref()
+ .ok_or_else(|| PyRuntimeError::new_err("producer has been shut down"))?;
+ producer
+ .send_to(stream, topic, messages, partitioning)
+ .await
+ .map(SendMessagesResponse::from)
+ .map_err(to_send_error)
+ })
+ }
+
+ #[gen_stub(skip)]
+ fn _is_send_active(&self) -> bool {
+ self.inner.try_write().is_err()
+ }
+
+ /// Waits for active sends and closes the producer. Repeated calls are safe.
+ /// Background shutdown flushes every accepted buffered message before returning.
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[None]", imports=("collections.abc")))]
+ fn shutdown<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ shutdown(inner).await?;
+ Ok(Python::attach(|py| py.None()))
+ })
+ }
+
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[IggyProducer]", imports=("collections.abc")))]
+ fn __aenter__<'py>(this: Py<Self>, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+ future_into_py(py, async move { Ok(this) })
+ }
+
+ #[gen_stub(override_return_type(type_repr = "collections.abc.Awaitable[builtins.bool]", imports=("collections.abc")))]
+ fn __aexit__<'py>(
+ &self,
+ py: Python<'py>,
+ _exc_type: &Bound<'_, PyAny>,
+ _exc_value: &Bound<'_, PyAny>,
+ _traceback: &Bound<'_, PyAny>,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let inner = self.inner.clone();
+ future_into_py(py, async move {
+ shutdown(inner).await?;
+ Ok(false)
+ })
+ }
+}
+
+#[derive(Clone, FromPyObject)]
+pub(crate) enum ProducerMode {
+ #[pyo3(transparent)]
+ Direct(DirectProducerConfig),
+ #[pyo3(transparent)]
+ Background(BackgroundProducerConfig),
+}
+
+impl Default for ProducerMode {
+ fn default() -> Self {
+ Self::Direct(DirectProducerConfig::default())
+ }
+}
+
+#[derive(FromPyObject)]
+#[pyo3(transparent)]
+struct DefaultBackpressureMode(BackpressureMode);
+
+impl Default for DefaultBackpressureMode {
+ fn default() -> Self {
+ Self(BackpressureMode::block())
+ }
+}
+
+impl DefaultBackpressureMode {
+ fn resolve(self) -> BackpressureMode {
+ self.0
+ }
+}
+
+impl PyStubType for DefaultBackpressureMode {
+ fn type_output() -> TypeInfo {
+ BackpressureMode::type_output()
+ }
+
+ fn type_input() -> TypeInfo {
+ BackpressureMode::type_input()
+ }
+}
+
+impl<'py> IntoPyObject<'py> for DefaultBackpressureMode {
+ type Target = PyAny;
+ type Output = Bound<'py, PyAny>;
+ type Error = PyErr;
+
+ fn into_pyobject(self, py: Python<'py>) -> Result<Self::Output, Self::Error> {
+ Ok(opaque_stub_default(py))
+ }
+}
+
+#[derive(FromPyObject)]
+#[pyo3(transparent)]
+struct DefaultProducerSharding(ProducerSharding);
+
+impl Default for DefaultProducerSharding {
+ fn default() -> Self {
+ Self(ProducerSharding::Ordered)
+ }
+}
+
+impl DefaultProducerSharding {
+ fn resolve(self) -> ProducerSharding {
+ self.0
+ }
+}
+
+impl PyStubType for DefaultProducerSharding {
+ fn type_output() -> TypeInfo {
+ ProducerSharding::type_output()
+ }
+
+ fn type_input() -> TypeInfo {
+ ProducerSharding::type_input()
+ }
+}
+
+impl<'py> IntoPyObject<'py> for DefaultProducerSharding {
+ type Target = PyAny;
+ type Output = Bound<'py, PyAny>;
+ type Error = PyErr;
+
+ fn into_pyobject(self, py: Python<'py>) -> Result<Self::Output, Self::Error> {
+ Ok(opaque_stub_default(py))
+ }
+}
+
+#[derive(Default)]
+pub(crate) enum RetryInterval {
+ #[default]
+ Omitted,
+ Disabled,
+ Duration(Py<PyDelta>),
+}
+
+impl<'a, 'py> FromPyObject<'a, 'py> for RetryInterval {
+ type Error = PyErr;
+
+ fn extract(object: Borrowed<'a, 'py, PyAny>) -> PyResult<Self> {
+ if object.is_none() {
+ Ok(Self::Disabled)
+ } else {
+ Ok(Self::Duration(object.extract::<Py<PyDelta>>()?))
+ }
+ }
+}
+
+impl PyStubType for RetryInterval {
+ fn type_output() -> TypeInfo {
+ <std::time::Duration>::type_output() | TypeInfo::none()
+ }
+
+ fn type_input() -> TypeInfo {
+ <std::time::Duration>::type_input() | TypeInfo::none()
+ }
+}
+
+impl<'py> IntoPyObject<'py> for RetryInterval {
+ type Target = PyAny;
+ type Output = Bound<'py, PyAny>;
+ type Error = PyErr;
+
+ fn into_pyobject(self, py: Python<'py>) -> Result<Self::Output, Self::Error> {
+ match self {
+ Self::Omitted => std::time::Duration::from_secs(1).into_bound_py_any(py),
+ Self::Disabled => Ok(py.None().into_bound(py)),
+ Self::Duration(duration) => Ok(duration.into_bound(py).into_any()),
+ }
+ }
+}
+
+impl RetryInterval {
+ pub(crate) fn resolve(self) -> PyResult<Option<iggy::prelude::NonZeroIggyDuration>> {
+ match self {
+ Self::Omitted => Ok(Some(iggy::prelude::NonZeroIggyDuration::ONE_SECOND)),
+ Self::Disabled => Ok(None),
+ Self::Duration(duration) => py_delta_to_iggy_duration(&duration).and_then(|duration| {
+ iggy::prelude::NonZeroIggyDuration::try_from(duration)
+ .map(Some)
+ .map_err(|_| PyValueError::new_err("'send_retry_interval' must not be zero"))
+ }),
+ }
+ }
+}
+
+#[derive(Default)]
+enum DefaultDuration {
+ #[default]
+ Zero,
+ OneMillisecond,
+ Value(Py<PyDelta>),
+}
+
+impl DefaultDuration {
+ fn one_millisecond() -> Self {
+ Self::OneMillisecond
+ }
+}
+
+impl PyStubType for DefaultDuration {
+ fn type_output() -> TypeInfo {
+ timedelta_type_info()
+ }
+
+ fn type_input() -> TypeInfo {
+ timedelta_type_info()
+ }
+}
+
+impl<'py> IntoPyObject<'py> for DefaultDuration {
+ type Target = PyDelta;
+ type Output = Bound<'py, PyDelta>;
+ type Error = PyErr;
+
+ fn into_pyobject(self, py: Python<'py>) -> Result<Self::Output, Self::Error> {
+ match self {
+ Self::Zero => std::time::Duration::ZERO.into_pyobject(py),
+ Self::OneMillisecond => std::time::Duration::from_millis(1).into_pyobject(py),
+ Self::Value(duration) => Ok(duration.into_bound(py)),
+ }
+ }
+}
+
+impl<'a, 'py> FromPyObject<'a, 'py> for DefaultDuration {
+ type Error = PyErr;
+
+ fn extract(object: Borrowed<'a, 'py, PyAny>) -> PyResult<Self> {
+ Ok(Self::Value(object.extract::<Py<PyDelta>>()?))
+ }
+}
+
+impl DefaultDuration {
+ fn resolve(self, default: IggyDuration) -> PyResult<IggyDuration> {
+ match self {
+ Self::Value(duration) => py_delta_to_iggy_duration(&duration),
+ Self::Zero | Self::OneMillisecond => Ok(default),
+ }
+ }
+}
+
+fn timedelta_type_info() -> TypeInfo {
+ let mut type_info = <std::time::Duration>::type_input();
+ type_info.source_module = None;
+ type_info
+}
+
+fn opaque_stub_default(py: Python<'_>) -> Bound<'_, PyAny> {
+ // The stub generator renders objects without a stable Python expression as
+ // `...`. An actual Ellipsis is rendered as `Ellipsis`, which type checkers
+ // reject as a default for these public configuration types.
+ py.None().into_bound(py).get_type().into_any()
+}
+
+async fn shutdown(inner: Arc<RwLock<Option<RustIggyProducer>>>) -> PyResult<()> {
+ // Exclusive access must span consuming shutdown so another close waits for
+ // completion and no send can observe a producer being closed underneath it.
+ let mut producer = inner.write().await;
+ if let Some(producer) = producer.take() {
+ producer.shutdown().await;
+ }
+ Ok(())
+}
+
+fn extract_messages(messages: &Bound<'_, PyList>) -> PyResult<Vec<RustIggyMessage>> {
+ messages
+ .iter()
+ .map(|item| {
+ let message = item.extract::<PyRef<'_, SendMessage>>()?;
+ Ok(clone_rust_message(&message))
+ })
+ .collect()
+}
+
+fn extract_send_to_identifier(value: &Bound<'_, PyAny>, parameter: &str) -> PyResult<Identifier> {
+ if let Ok(value) = value.cast::<PyString>() {
+ return Identifier::from_str(value.to_str()?)
+ .map_err(|error| PyValueError::new_err(error.to_string()));
+ }
+ if value.is_instance_of::<PyInt>() {
+ let value = value.extract::<u32>()?;
+ return Identifier::numeric(value)
+ .map_err(|error| PyValueError::new_err(error.to_string()));
+ }
+ Err(PyTypeError::new_err(format!(
+ "'{parameter}' must be a string or an integer"
+ )))
+}
+
+fn clone_rust_message(message: &SendMessage) -> RustIggyMessage {
+ message.clone().inner
+}
+
+fn to_send_error(error: IggyError) -> PyErr {
+ match error {
+ IggyError::ProducerSendFailed {
+ cause,
+ failed,
+ committed,
+ ..
+ } => ProducerSendError::new_err(*cause, failed, committed),
+ error => PyRuntimeError::new_err(error.to_string()),
+ }
+}
+
+pub(crate) fn u32_param(value: i64, parameter: &str) -> PyResult<u32> {
+ u32::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u32::MAX))
+ })
+}
+
+fn usize_param(value: i128, parameter: &str) -> PyResult<usize> {
+ usize::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!(
+ "'{parameter}' must be between 0 and {}",
+ usize::MAX
+ ))
+ })
+}
+
+fn u64_param(value: i128, parameter: &str) -> PyResult<u64> {
+ u64::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!("'{parameter}' must be between 0 and {}", u64::MAX))
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use std::future::Future;
+ use std::sync::atomic::{AtomicUsize, Ordering};
+ use std::time::Duration;
+
+ use iggy::clients::producer::ProducerCoreBackend;
+ use iggy::clients::producer_dispatcher::ProducerDispatcher;
+ use pyo3::exceptions::{PyOverflowError, PyRuntimeError, PyTypeError, PyValueError};
+
+ use super::*;
+
+ #[derive(Debug)]
+ struct ConcurrencyTrackingBackend {
+ active: Arc<AtomicUsize>,
+ maximum: Arc<AtomicUsize>,
+ }
+
+ impl ProducerCoreBackend for ConcurrencyTrackingBackend {
+ fn send_internal(
+ &self,
+ _stream: &Identifier,
+ _topic: &Identifier,
+ _messages: Vec<RustIggyMessage>,
+ _partitioning: Option<Arc<iggy::prelude::Partitioning>>,
+ ) -> impl Future<Output = Result<iggy::prelude::SendMessagesResponse, IggyError>> + Send
+ {
+ let active = self.active.clone();
+ let maximum = self.maximum.clone();
+ async move {
+ let active_now = active.fetch_add(1, Ordering::SeqCst) + 1;
+ maximum.fetch_max(active_now, Ordering::SeqCst);
+ tokio::time::sleep(Duration::from_millis(50)).await;
+ active.fetch_sub(1, Ordering::SeqCst);
+ Ok(iggy::prelude::SendMessagesResponse {
+ confirmations: Vec::new(),
+ })
+ }
+ }
+ }
+
+ #[test]
+ fn direct_defaults_match_rust() {
+ let python = DirectProducerConfig::default();
+ let rust = RustDirectConfig::builder().build();
+
+ assert_eq!(python.inner.batch_length, rust.batch_length);
+ assert_eq!(python.inner.linger_time, rust.linger_time);
+ }
+
+ #[test]
+ fn direct_linger_rejects_values_above_u64_microseconds() {
+ Python::initialize();
+ Python::attach(|py| {
+ let duration = PyDelta::new(py, 999_999_999, 0, 0, false).unwrap().unbind();
+ let error = DirectProducerConfig::new(1_000, DefaultDuration::Value(duration))
+ .err()
+ .unwrap();
+
+ assert!(error.is_instance_of::<PyValueError>(py));
+ assert!(error.to_string().contains("linger_time"));
+ });
+ }
+
+ #[test]
+ fn producer_send_error_preserves_recovery_state() {
+ Python::initialize();
+ Python::attach(|py| {
+ let cause = IggyError::CannotSendMessagesDueToClientDisconnection;
+ let cause_message = cause.to_string();
+ let error = to_send_error(IggyError::ProducerSendFailed {
+ cause: Box::new(cause),
+ failed: Arc::new(vec![RustIggyMessage::from_str("failed").unwrap()]),
+ committed: Arc::new(vec![RustSendMessagesConfirmationResponse {
+ stream_id: 1,
+ topic_id: 2,
+ partition_id: 3,
+ base_offset: 4,
+ }]),
+ stream_name: "stream".to_owned(),
+ topic_name: "topic".to_owned(),
+ });
+
+ assert!(error.is_instance_of::<ProducerSendError>(py));
+ assert!(error.is_instance_of::<PyRuntimeError>(py));
+ assert_eq!(
+ error
+ .value(py)
+ .getattr("cause")
+ .unwrap()
+ .extract::<String>()
+ .unwrap(),
+ cause_message
+ );
+
+ let failed = error
+ .value(py)
+ .getattr("failed")
+ .unwrap()
+ .extract::<Vec<Py<SendMessage>>>()
+ .unwrap();
+ assert_eq!(failed.len(), 1);
+ assert_eq!(failed[0].borrow(py).inner.payload.as_ref(), b"failed");
+
+ let committed = error
+ .value(py)
+ .getattr("committed")
+ .unwrap()
+ .extract::<Vec<Py<SendMessagesConfirmation>>>()
+ .unwrap();
+ assert_eq!(committed.len(), 1);
+ assert_eq!(committed[0].borrow(py).inner.base_offset, 4);
+
+ let source = error.cause(py).unwrap();
+ assert!(source.is_instance_of::<PyRuntimeError>(py));
+ assert_eq!(
+ source.value(py).str().unwrap().to_str().unwrap(),
+ cause_message
+ );
+ });
+ }
+
+ #[test]
+ fn background_defaults_match_rust() {
+ let python = BackgroundProducerConfig::default();
+ let rust = RustBackgroundConfig::builder().build();
+
+ assert_eq!(python.num_shards, rust.num_shards);
+ assert_eq!(python.linger_time, rust.linger_time);
+ assert_eq!(python.batch_size, rust.batch_size);
+ assert_eq!(python.batch_length, rust.batch_length);
+ assert_eq!(python.max_buffer_size, rust.max_buffer_size);
+ assert!(matches!(rust.failure_mode, RustBackpressureMode::Block));
+ assert_eq!(python.max_in_flight, rust.max_in_flight);
+ assert!(matches!(python.sharding, ProducerSharding::Ordered));
+ assert_eq!(format!("{:?}", rust.sharding), "OrderedSharding");
+ }
+
+ #[test]
+ fn background_configuration_converts_every_field_to_rust() {
+ let python = BackgroundProducerConfig {
+ num_shards: 4,
+ linger_time: IggyDuration::from(2_000),
+ batch_size: 2_048,
+ batch_length: 32,
+ max_buffer_size: IggyByteSize::from(8_192),
+ failure_mode: BackpressureMode::fail_immediately(),
+ max_in_flight: 3,
+ sharding: ProducerSharding::Balanced,
+ };
+ let rust = RustBackgroundConfig::try_from(&python).unwrap();
+
+ assert_eq!(rust.num_shards, 4);
+ assert_eq!(rust.linger_time, IggyDuration::from(2_000));
+ assert_eq!(rust.batch_size, 2_048);
+ assert_eq!(rust.batch_length, 32);
+ assert_eq!(rust.max_buffer_size.as_bytes_u64(), 8_192);
+ assert!(matches!(
+ rust.failure_mode,
+ RustBackpressureMode::FailImmediately
+ ));
+ assert_eq!(rust.max_in_flight, 3);
+ assert_eq!(
+ format!("{:?}", rust.sharding),
+ "BalancedSharding { counter: 0 }"
+ );
+ }
+
+ #[test]
+ fn backpressure_modes_convert_to_rust() {
+ let block = RustBackpressureMode::from(&BackpressureMode::block());
+ let timeout = RustBackpressureMode::from(&BackpressureMode {
+ kind: BackpressureKind::BlockWithTimeout(IggyDuration::from(250_000)),
+ });
+ let immediate = RustBackpressureMode::from(&BackpressureMode::fail_immediately());
+
+ assert!(matches!(block, RustBackpressureMode::Block));
+ assert!(matches!(
+ timeout,
+ RustBackpressureMode::BlockWithTimeout(value)
+ if value == IggyDuration::from(250_000)
+ ));
+ assert!(matches!(immediate, RustBackpressureMode::FailImmediately));
+ }
+
+ #[test]
+ fn sharding_modes_convert_to_the_selected_rust_strategy() {
+ let messages = vec![RustIggyMessage::from_str("message").unwrap()];
+ let stream = Identifier::from_str("stream").unwrap();
+ let topic = Identifier::from_str("topic").unwrap();
+
+ let ordered = RustBackgroundConfig::try_from(&BackgroundProducerConfig {
+ num_shards: 4,
+ ..BackgroundProducerConfig::default()
+ })
+ .unwrap();
+ let first = ordered.sharding.pick_shard(4, &messages, &stream, &topic);
+ let second = ordered.sharding.pick_shard(4, &messages, &stream, &topic);
+ assert_eq!(first, second);
+
+ let balanced = RustBackgroundConfig::try_from(&BackgroundProducerConfig {
+ num_shards: 3,
+ sharding: ProducerSharding::Balanced,
+ ..BackgroundProducerConfig::default()
+ })
+ .unwrap();
+ let picked = (0..4)
+ .map(|_| balanced.sharding.pick_shard(3, &messages, &stream, &topic))
+ .collect::<Vec<_>>();
+ assert_eq!(picked, vec![0, 1, 2, 0]);
+ }
+
+ #[test]
+ fn background_conversion_rejects_values_that_would_panic_semaphore() {
+ let invalid_max_buffer = BackgroundProducerConfig {
+ max_buffer_size: IggyByteSize::from(Semaphore::MAX_PERMITS as u64 + 1),
+ ..BackgroundProducerConfig::default()
+ };
+ let invalid_max_in_flight = BackgroundProducerConfig {
+ max_in_flight: Semaphore::MAX_PERMITS + 1,
+ ..BackgroundProducerConfig::default()
+ };
+
+ assert!(RustBackgroundConfig::try_from(&invalid_max_buffer).is_err());
+ assert!(RustBackgroundConfig::try_from(&invalid_max_in_flight).is_err());
+ }
+
+ #[tokio::test]
+ async fn converted_max_in_flight_limits_concurrent_background_writes() {
+ let active = Arc::new(AtomicUsize::new(0));
+ let maximum = Arc::new(AtomicUsize::new(0));
+ let backend = Arc::new(ConcurrencyTrackingBackend {
+ active,
+ maximum: maximum.clone(),
+ });
+ let config = RustBackgroundConfig::try_from(&BackgroundProducerConfig {
+ num_shards: 4,
+ linger_time: IggyDuration::from(0),
+ batch_size: 0,
+ batch_length: 1,
+ max_buffer_size: IggyByteSize::from(0),
+ failure_mode: BackpressureMode::block(),
+ max_in_flight: 2,
+ sharding: ProducerSharding::Balanced,
+ })
+ .unwrap();
+ let dispatcher = ProducerDispatcher::new(backend, config);
+ let stream = Arc::new(Identifier::numeric(1).unwrap());
+ let topic = Arc::new(Identifier::numeric(1).unwrap());
+
+ for index in 0..4 {
+ dispatcher
+ .dispatch(
+ vec![RustIggyMessage::from_str(&index.to_string()).unwrap()],
+ stream.clone(),
+ topic.clone(),
+ None,
+ )
+ .await
+ .unwrap();
+ }
+ dispatcher.shutdown().await;
+
+ assert_eq!(maximum.load(Ordering::SeqCst), 2);
+ }
+
+ #[test]
+ fn direct_batch_length_rejects_values_outside_u32() {
+ assert!(u32_param(-1, "batch_length").is_err());
+ assert!(u32_param(i64::from(u32::MAX) + 1, "batch_length").is_err());
+ assert_eq!(
+ u32_param(i64::from(u32::MAX), "batch_length").unwrap(),
+ u32::MAX
+ );
+ }
+
+ #[test]
+ fn send_to_identifier_preserves_python_error_categories() {
+ Python::initialize();
+ Python::attach(|py| {
+ let string = PyString::new(py, "stream");
+ let numeric = 7_u32.into_pyobject(py).unwrap();
+ let negative = (-1_i64).into_pyobject(py).unwrap();
+ let too_large = (u64::from(u32::MAX) + 1).into_pyobject(py).unwrap();
+ let wrong_type = py.None().into_bound(py);
+
+ assert_eq!(
+ extract_send_to_identifier(string.as_any(), "stream")
+ .unwrap()
+ .get_string_value()
+ .unwrap(),
+ "stream"
+ );
+ assert_eq!(
+ extract_send_to_identifier(numeric.as_any(), "stream")
+ .unwrap()
+ .get_u32_value()
+ .unwrap(),
+ 7
+ );
+
+ let negative = extract_send_to_identifier(negative.as_any(), "stream").unwrap_err();
+ let too_large = extract_send_to_identifier(too_large.as_any(), "stream").unwrap_err();
+ let wrong_type = extract_send_to_identifier(&wrong_type, "stream").unwrap_err();
+
+ assert!(negative.is_instance_of::<PyOverflowError>(py));
+ assert!(too_large.is_instance_of::<PyOverflowError>(py));
+ assert!(wrong_type.is_instance_of::<PyTypeError>(py));
+ });
+ }
+
+ #[test]
+ fn representations_name_public_python_constructors() {
+ let direct = DirectProducerConfig::default();
+ let background = BackgroundProducerConfig::default();
+
+ assert_eq!(
+ direct.__repr__(),
+ "DirectProducerConfig(batch_length=1000, linger_time=datetime.timedelta(seconds=0))"
+ );
+ assert_eq!(
+ BackpressureMode::block().__repr__(),
+ "BackpressureMode.block()"
+ );
+ assert_eq!(
+ BackpressureMode::fail_immediately().__repr__(),
+ "BackpressureMode.fail_immediately()"
+ );
+ assert!(
+ background
+ .__repr__()
+ .ends_with("sharding=ProducerSharding.ORDERED)")
+ );
+ }
+
+ #[test]
+ fn numeric_getters_return_configured_values() {
+ let background = BackgroundProducerConfig::default();
+
+ assert_eq!(background.num_shards(), 1);
+ assert_eq!(background.batch_size(), 1_048_576);
+ assert_eq!(background.batch_length(), 1_000);
+ assert_eq!(background.max_buffer_size(), 33_554_432);
+ assert_eq!(background.max_in_flight(), 1);
+ assert_eq!(background.sharding(), ProducerSharding::Ordered);
+ assert_eq!(background.failure_mode(), BackpressureMode::block());
+ }
+
+ #[test]
+ fn numeric_validation_uses_python_semantic_ranges() {
+ assert!(usize_param(-1, "num_shards").is_err());
+ assert!(u64_param(-1, "max_buffer_size").is_err());
+ assert_eq!(
+ usize_param(usize::MAX as i128, "num_shards").unwrap(),
+ usize::MAX
+ );
+ assert_eq!(
+ u64_param(u64::MAX as i128, "max_buffer_size").unwrap(),
+ u64::MAX
+ );
+ }
+
+ #[test]
+ fn direct_default_constants_match_builder() {
+ let rust = RustDirectConfig::builder().build();
+ assert_eq!(1_000, rust.batch_length);
+ assert_eq!(IggyDuration::from(0), rust.linger_time);
+ }
+
+ #[test]
+ fn background_default_constants_match_builder() {
+ let rust = RustBackgroundConfig::builder().build();
+ assert_eq!(DEFAULT_BACKGROUND_NUM_SHARDS, rust.num_shards);
+ assert_eq!(IggyDuration::from(1_000), rust.linger_time);
+ assert_eq!(DEFAULT_BACKGROUND_BATCH_SIZE, rust.batch_size);
+ assert_eq!(DEFAULT_BACKGROUND_BATCH_LENGTH, rust.batch_length);
+ assert_eq!(
+ DEFAULT_BACKGROUND_MAX_BUFFER_SIZE,
+ rust.max_buffer_size.as_bytes_u64()
+ );
+ assert_eq!(DEFAULT_BACKGROUND_MAX_IN_FLIGHT, rust.max_in_flight);
+ }
+}
diff --git a/foreign/python/src/send_message.rs b/foreign/python/src/send_message.rs
index 8cb940fc4..454fbb2ac 100644
--- a/foreign/python/src/send_message.rs
+++ b/foreign/python/src/send_message.rs
@@ -38,20 +38,26 @@ pub struct SendMessage {
impl Clone for SendMessage {
fn clone(&self) -> Self {
+ Self::clone_from_rust(&self.inner)
+ }
+}
+
+impl SendMessage {
+ pub(crate) fn clone_from_rust(message: &RustIggyMessage) -> Self {
Self {
inner: RustIggyMessage {
header: IggyMessageHeader {
- checksum: self.inner.header.checksum,
- id: self.inner.header.id,
- offset: self.inner.header.offset,
- timestamp: self.inner.header.timestamp,
- origin_timestamp: self.inner.header.origin_timestamp,
- user_headers_length: self.inner.header.user_headers_length,
- payload_length: self.inner.header.payload_length,
- reserved: self.inner.header.reserved,
+ checksum: message.header.checksum,
+ id: message.header.id,
+ offset: message.header.offset,
+ timestamp: message.header.timestamp,
+ origin_timestamp: message.header.origin_timestamp,
+ user_headers_length: message.header.user_headers_length,
+ payload_length: message.header.payload_length,
+ reserved: message.header.reserved,
},
- payload: self.inner.payload.clone(),
- user_headers: self.inner.user_headers.clone(),
+ payload: message.payload.clone(),
+ user_headers: message.user_headers.clone(),
},
}
}
diff --git a/foreign/python/tests/test_producer.py b/foreign/python/tests/test_producer.py
new file mode 100644
index 000000000..418b7631e
--- /dev/null
+++ b/foreign/python/tests/test_producer.py
@@ -0,0 +1,1608 @@
+# 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.
+
+"""Tests for the high-level producer API and both send modes."""
+
+import ast
+import asyncio
+import contextlib
+import os
+import subprocess
+import sys
+import time
+from datetime import timedelta
+from pathlib import Path
+
+import pytest
+
+from apache_iggy import (
+ BackgroundProducerConfig,
+ BackpressureMode,
+ Consumer,
+ DirectProducerConfig,
+ IggyClient,
+ IggyExpiry,
+ IggyProducer,
+ MaxTopicSize,
+ Partitioning,
+ PollingStrategy,
+ ProducerSendError,
+ ProducerSharding,
+ SendMessage,
+ SendMessagesResponse,
+)
+
+from .utils import wait_for_server
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[3]
+SERVER_BINARY = (
+ REPOSITORY_ROOT
+ / "target"
+ / "debug"
+ / ("iggy-server.exe" if sys.platform == "win32" else "iggy-server")
+)
+
+
+def spawn_restartable_server(
+ data_path: Path,
+ tcp_address: str,
+) -> subprocess.Popen[bytes]:
+ data_path.mkdir(parents=True, exist_ok=True)
+ environment = os.environ.copy()
+ # Python test connection settings use the IGGY_SERVER_ prefix, but the
+ # server rejects them as unknown configuration overrides.
+ for name in tuple(environment):
+ if name.startswith("IGGY_SERVER_"):
+ del environment[name]
+ environment.update(
+ {
+ "IGGY_PATH": str(data_path),
+ "IGGY_TCP_ADDRESS": tcp_address,
+ "IGGY_HTTP_ENABLED": "false",
+ "IGGY_QUIC_ENABLED": "false",
+ "IGGY_WEBSOCKET_ENABLED": "false",
+ "IGGY_LOGGING_LEVEL": "error",
+ }
+ )
+ with (data_path / "server.log").open("ab") as server_log:
+ return subprocess.Popen( # noqa: S603
+ [str(SERVER_BINARY), "--with-default-root-credentials"],
+ cwd=REPOSITORY_ROOT,
+ env=environment,
+ stdout=server_log,
+ stderr=subprocess.STDOUT,
+ )
+
+
+async def stop_restartable_server(process: subprocess.Popen[bytes]) -> None:
+ if process.poll() is not None:
+ return
+
+ process.terminate()
+ try:
+ await asyncio.to_thread(process.wait, 10)
+ except subprocess.TimeoutExpired:
+ process.kill()
+ await asyncio.to_thread(process.wait)
+
+
+async def discover_server_address(
+ data_path: Path,
+ process: subprocess.Popen[bytes],
+ timeout: float = 15,
+) -> tuple[str, int]:
+ config_path = data_path / "runtime" / "current_config.toml"
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ if process.poll() is not None:
+ server_log = (data_path / "server.log").read_text(errors="replace")
+ raise RuntimeError(
+ f"Restartable Iggy server exited with code {process.returncode}:\n"
+ f"{server_log[-4_000:]}"
+ )
+ if config_path.exists():
+ in_tcp_section = False
+ for line in config_path.read_text().splitlines():
+ stripped = line.strip()
+ if stripped.startswith("["):
+ in_tcp_section = stripped == "[tcp]"
+ elif in_tcp_section and stripped.startswith("address = "):
+ address = stripped.split('"', 2)[1]
+ host, port = address.rsplit(":", 1)
+ return host, int(port)
+ await asyncio.sleep(0.05)
+
+ raise TimeoutError("Restartable Iggy server did not publish its TCP address")
+
+
+async def poll_payloads(
+ client: IggyClient,
+ stream: str,
+ topic: str,
+ *,
+ partition_id: int = 0,
+) -> list[str]:
+ messages = await client.poll_messages(
+ stream=stream,
+ topic=topic,
+ consumer=Consumer.Single(1),
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.First(),
+ count=100,
+ auto_commit=False,
+ )
+ return [message.payload().decode("utf-8") for message in messages]
+
+
+async def wait_for_payloads(
+ client: IggyClient,
+ stream: str,
+ topic: str,
+ expected: list[str],
+ *,
+ partition_id: int = 0,
+ timeout: float = 2,
+) -> list[str]:
+ deadline = time.monotonic() + timeout
+ payloads: list[str] = []
+ while time.monotonic() < deadline:
+ payloads = await poll_payloads(
+ client,
+ stream,
+ topic,
+ partition_id=partition_id,
+ )
+ if payloads == expected:
+ return payloads
+ await asyncio.sleep(0.01)
+ raise AssertionError(
+ f"Timed out waiting for payloads {expected!r}; last payloads were {payloads!r}"
+ )
+
+
+@pytest.mark.unit
+class TestDirectProducerConfig:
+ """Test direct-mode configuration independently of a server."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ config = DirectProducerConfig()
+
+ assert config.batch_length == 1_000
+ assert config.linger_time == timedelta(0)
+
+ def test_fields_are_keyword_only_and_round_trip(self):
+ config = DirectProducerConfig(
+ batch_length=25,
+ linger_time=timedelta(milliseconds=125),
+ )
+
+ assert config.batch_length == 25
+ assert config.linger_time == timedelta(milliseconds=125)
+
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ DirectProducerConfig(25, timedelta(milliseconds=125))
+
+ def test_zero_batch_length_retains_the_rust_sentinel(self):
+ assert DirectProducerConfig(batch_length=0).batch_length == 0
+
+ def test_largest_batch_length_round_trips(self):
+ assert DirectProducerConfig(batch_length=2**32 - 1).batch_length == 2**32 - 1
+
+ @pytest.mark.parametrize("batch_length", [-1, 2**32])
+ def test_batch_length_outside_u32_is_rejected(self, batch_length: int):
+ with pytest.raises(ValueError, match="batch_length"):
+ DirectProducerConfig(batch_length=batch_length)
+
+ def test_batch_length_binding_overflow_is_distinct(self):
+ with pytest.raises(OverflowError):
+ DirectProducerConfig(batch_length=2**63)
+
+ @pytest.mark.parametrize(
+ "linger_time",
+ [timedelta(microseconds=-1), timedelta(seconds=-1)],
+ )
+ def test_negative_linger_is_rejected(self, linger_time: timedelta):
+ with pytest.raises(ValueError, match="negative"):
+ DirectProducerConfig(linger_time=linger_time)
+
+ def test_linger_above_u64_microseconds_is_rejected(self):
+ with pytest.raises(ValueError, match="linger_time"):
+ DirectProducerConfig(linger_time=timedelta(days=999_999_999))
+
+ def test_wrong_field_types_are_rejected(self):
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ DirectProducerConfig(batch_length="1000")
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ DirectProducerConfig(linger_time=1)
+
+ def test_repr_contains_pasteable_python_values(self):
+ config = DirectProducerConfig(
+ batch_length=25,
+ linger_time=timedelta(milliseconds=125),
+ )
+
+ printed = repr(config)
+
+ assert "batch_length=25" in printed
+ assert "linger_time=datetime.timedelta(microseconds=125000)" in printed
+ ast.parse(printed)
+
+
+@pytest.mark.unit
+class TestBackgroundProducerSurface:
+ """Test the background configuration model independently of a server."""
+
+ def test_sharding_values_are_public_and_distinct(self):
+ assert ProducerSharding.ORDERED != ProducerSharding.BALANCED
+ assert repr(ProducerSharding.ORDERED) == "ProducerSharding.ORDERED"
+ assert repr(ProducerSharding.BALANCED) == "ProducerSharding.BALANCED"
+
+ def test_backpressure_constructors_and_getters(self):
+ blocking = BackpressureMode.block()
+ timed = BackpressureMode.block_with_timeout(timedelta(milliseconds=250))
+ immediate = BackpressureMode.fail_immediately()
+
+ assert blocking.timeout is None
+ assert timed.timeout == timedelta(milliseconds=250)
+ assert immediate.timeout is None
+ assert repr(blocking) == "BackpressureMode.block()"
+ assert repr(timed) == (
+ "BackpressureMode.block_with_timeout("
+ "datetime.timedelta(microseconds=250000))"
+ )
+ assert repr(immediate) == "BackpressureMode.fail_immediately()"
+
+ def test_backpressure_timeout_validation(self):
+ zero = BackpressureMode.block_with_timeout(timedelta(0))
+ assert zero.timeout == timedelta(0)
+
+ with pytest.raises(ValueError, match="negative"):
+ BackpressureMode.block_with_timeout(timedelta(microseconds=-1))
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ BackpressureMode.block_with_timeout(1)
+
+ def test_defaults_match_the_rust_sdk(self):
+ config = BackgroundProducerConfig()
+
+ assert config.num_shards == 1
+ assert config.linger_time == timedelta(milliseconds=1)
+ assert config.batch_size == 1024 * 1024
+ assert config.batch_length == 1_000
+ assert config.max_buffer_size == 32 * 1024 * 1024
+ assert repr(config.failure_mode) == "BackpressureMode.block()"
+ assert config.max_in_flight == 1
+ assert config.sharding == ProducerSharding.ORDERED
+
+ def test_every_field_round_trips(self):
+ failure_mode = BackpressureMode.block_with_timeout(timedelta(milliseconds=500))
+ config = BackgroundProducerConfig(
+ num_shards=4,
+ linger_time=timedelta(milliseconds=10),
+ batch_size=256 * 1024,
+ batch_length=50,
+ max_buffer_size=8 * 1024 * 1024,
+ failure_mode=failure_mode,
+ max_in_flight=3,
+ sharding=ProducerSharding.BALANCED,
+ )
+
+ assert config.num_shards == 4
+ assert config.linger_time == timedelta(milliseconds=10)
+ assert config.batch_size == 256 * 1024
+ assert config.batch_length == 50
+ assert config.max_buffer_size == 8 * 1024 * 1024
+ assert repr(config.failure_mode) == repr(failure_mode)
+ assert config.max_in_flight == 3
+ assert config.sharding == ProducerSharding.BALANCED
+
+ def test_fields_are_keyword_only_and_read_only(self):
+ config = BackgroundProducerConfig()
+
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ BackgroundProducerConfig(2)
+ with pytest.raises(AttributeError):
+ # pyrefly: ignore # read-only
+ config.num_shards = 2
+
+ @pytest.mark.parametrize(
+ "field",
+ [
+ "num_shards",
+ "batch_size",
+ "batch_length",
+ "max_buffer_size",
+ "max_in_flight",
+ ],
+ )
+ def test_unsigned_fields_reject_negative_values(self, field: str):
+ with pytest.raises(ValueError, match=field):
+ # pyrefly: ignore # bad-argument-type
+ BackgroundProducerConfig(**{field: -1})
+
+ @pytest.mark.parametrize(
+ "field",
+ [
+ "num_shards",
+ "batch_size",
+ "batch_length",
+ "max_buffer_size",
+ "max_in_flight",
+ ],
+ )
+ def test_unsigned_fields_reject_wrong_types(self, field: str):
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ BackgroundProducerConfig(**{field: "1"})
+
+ def test_max_buffer_size_checks_semantic_range_before_binding_overflow(self):
+ maximum = 2**64 - 1
+
+ assert (
+ BackgroundProducerConfig(max_buffer_size=maximum).max_buffer_size == maximum
+ )
+ with pytest.raises(ValueError, match="max_buffer_size"):
+ BackgroundProducerConfig(max_buffer_size=2**64)
+ with pytest.raises(OverflowError):
+ BackgroundProducerConfig(max_buffer_size=2**127)
+
+ @pytest.mark.parametrize("value", [0, 1])
+ def test_zero_and_one_are_preserved_for_threshold_fields(self, value: int):
+ config = BackgroundProducerConfig(
+ num_shards=value,
+ batch_size=value,
+ batch_length=value,
+ max_buffer_size=value,
+ max_in_flight=value,
+ )
+
+ assert config.num_shards == value
+ assert config.batch_size == value
+ assert config.batch_length == value
+ assert config.max_buffer_size == value
+ assert config.max_in_flight == value
+
+ def test_wrong_variant_and_duration_types_are_rejected(self):
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ BackgroundProducerConfig(linger_time=1)
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ BackgroundProducerConfig(failure_mode="block")
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ BackgroundProducerConfig(sharding="ordered")
+
+ def test_negative_linger_is_rejected(self):
+ with pytest.raises(ValueError, match="negative"):
+ BackgroundProducerConfig(linger_time=timedelta(microseconds=-1))
+
+ def test_repr_contains_every_public_field(self):
+ config = BackgroundProducerConfig(
+ num_shards=2,
+ linger_time=timedelta(milliseconds=5),
+ batch_size=4_096,
+ batch_length=8,
+ max_buffer_size=65_536,
+ failure_mode=BackpressureMode.fail_immediately(),
+ max_in_flight=2,
+ sharding=ProducerSharding.BALANCED,
+ )
+
+ printed = repr(config)
+
+ for expected in [
+ "num_shards=2",
+ "linger_time=datetime.timedelta(microseconds=5000)",
+ "batch_size=4096",
+ "batch_length=8",
+ "max_buffer_size=65536",
+ "failure_mode=BackpressureMode.fail_immediately()",
+ "max_in_flight=2",
+ "sharding=ProducerSharding.BALANCED",
+ ]:
+ assert expected in printed
+ ast.parse(printed)
+
+
+class TestProducerCreation:
+ """Test initialization and producer-owned resource creation."""
+
+ @pytest.mark.asyncio
+ async def test_default_producer_creates_resources_and_is_immediately_usable(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+
+ producer = await iggy_client.producer(stream_name, topic_name)
+ try:
+ response = await producer.send_one(SendMessage("ready"))
+
+ assert isinstance(producer, IggyProducer)
+ assert isinstance(response, SendMessagesResponse)
+ assert len(response.confirmations) == 1
+ assert await iggy_client.get_stream(stream_name) is not None
+ assert await iggy_client.get_topic(stream_name, topic_name) is not None
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_none_mode_uses_the_default_direct_mode(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ mode=None,
+ )
+ try:
+ response = await producer.send_one(SendMessage("default mode"))
+ assert len(response.confirmations) == 1
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_existing_resources_work_when_creation_is_disabled(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(stream_name, topic_name, 1)
+
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ create_stream_if_not_exists=False,
+ create_topic_if_not_exists=False,
+ )
+ try:
+ response = await producer.send_one(SendMessage("existing"))
+ assert len(response.confirmations) == 1
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_missing_stream_fails_without_returning_a_producer(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ not_returned = object()
+ result = not_returned
+
+ with pytest.raises(RuntimeError):
+ result = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ create_stream_if_not_exists=False,
+ )
+
+ assert result is not_returned
+ assert await iggy_client.get_stream(stream_name) is None
+
+ @pytest.mark.asyncio
+ async def test_missing_topic_fails_after_creating_the_bound_stream(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ not_returned = object()
+ result = not_returned
+
+ with pytest.raises(RuntimeError):
+ result = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ create_topic_if_not_exists=False,
+ )
+
+ assert result is not_returned
+ assert await iggy_client.get_stream(stream_name) is not None
+ assert await iggy_client.get_topic(stream_name, topic_name) is None
+
+ @pytest.mark.asyncio
+ async def test_created_topic_uses_requested_topology_and_retention(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ expiry = timedelta(minutes=15)
+ maximum_size = 2_000_000_000
+
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ topic_partitions_count=3,
+ topic_message_expiry=IggyExpiry.ExpireDuration(expiry),
+ topic_max_size=MaxTopicSize.Custom(maximum_size),
+ )
+ try:
+ topic = await iggy_client.get_topic(stream_name, topic_name)
+
+ assert topic is not None
+ assert topic.partitions_count == 3
+ assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration)
+ assert topic.message_expiry.duration == expiry
+ assert isinstance(topic.max_topic_size, MaxTopicSize.Custom)
+ assert topic.max_topic_size.bytes == maximum_size
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_background_producer_is_initialized_and_immediately_usable(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("background-", min_bytes=20, max_bytes=20)
+ topic_name = unique_name("topic-", min_bytes=20, max_bytes=20)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(),
+ )
+ try:
+ response = await producer.send_one(SendMessage("ready"))
+
+ assert isinstance(producer, IggyProducer)
+ assert response.confirmations == []
+ finally:
+ await producer.shutdown()
+
+ assert await wait_for_payloads(
+ iggy_client, stream_name, topic_name, ["ready"]
+ ) == ["ready"]
+
+
+class TestProducerSends:
+ """Test all direct send operations and partitioning fallbacks."""
+
+ @pytest.mark.asyncio
+ async def test_send_and_send_one_return_confirmations(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+ try:
+ batch = await producer.send([SendMessage("first"), SendMessage("second")])
+ single = await producer.send_one(SendMessage("third"))
+
+ assert isinstance(batch, SendMessagesResponse)
+ assert isinstance(single, SendMessagesResponse)
+ assert len(batch.confirmations) == 1
+ assert len(single.confirmations) == 1
+ assert single.confirmations[0].base_offset == 2
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_default_producer_partitioning_is_balanced(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ topic_partitions_count=3,
+ )
+ try:
+ responses = [
+ await producer.send_one(SendMessage(str(index))) for index in range(3)
+ ]
+
+ assert {
+ response.confirmations[0].partition_id for response in responses
+ } == {0, 1, 2}
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_per_call_partitioning_overrides_and_falls_back(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ partitioning=Partitioning.partition_id(2),
+ topic_partitions_count=3,
+ )
+ try:
+ configured = await producer.send([SendMessage("configured")])
+ overridden = await producer.send_with_partitioning(
+ [SendMessage("override")], Partitioning.partition_id(0)
+ )
+ omitted = await producer.send_with_partitioning([SendMessage("omitted")])
+ explicit_none = await producer.send_with_partitioning(
+ [SendMessage("none")], None
+ )
+
+ assert configured.confirmations[0].partition_id == 2
+ assert overridden.confirmations[0].partition_id == 0
+ assert omitted.confirmations[0].partition_id == 2
+ assert explicit_none.confirmations[0].partition_id == 2
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_send_to_accepts_string_and_numeric_identifiers(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ destination_stream = unique_name()
+ destination_topic = unique_name()
+ await iggy_client.create_stream(destination_stream)
+ await iggy_client.create_topic(destination_stream, destination_topic, 2)
+ stream = await iggy_client.get_stream(destination_stream)
+ topic = await iggy_client.get_topic(destination_stream, destination_topic)
+ assert stream is not None
+ assert topic is not None
+
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ partitioning=Partitioning.partition_id(1),
+ )
+ try:
+ by_name = await producer.send_to(
+ destination_stream,
+ destination_topic,
+ [SendMessage("names")],
+ )
+ by_id = await producer.send_to(
+ stream.id,
+ topic.id,
+ [SendMessage("ids")],
+ None,
+ )
+
+ assert by_name.confirmations[0].stream_id == stream.id
+ assert by_name.confirmations[0].topic_id == topic.id
+ assert by_name.confirmations[0].partition_id == 1
+ assert by_id.confirmations[0].stream_id == stream.id
+ assert by_id.confirmations[0].topic_id == topic.id
+ assert by_id.confirmations[0].partition_id == 1
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_empty_batches_are_successful_no_ops_before_shutdown(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ producer = await iggy_client.producer(stream_name, topic_name)
+ try:
+ responses = [
+ await producer.send([]),
+ await producer.send_with_partitioning([]),
+ await producer.send_to(stream_name, topic_name, []),
+ ]
+
+ assert all(
+ isinstance(response, SendMessagesResponse) for response in responses
+ )
+ assert all(response.confirmations == [] for response in responses)
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_batch_length_splits_direct_requests(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ mode=DirectProducerConfig(batch_length=2),
+ )
+ try:
+ response = await producer.send(
+ [SendMessage(str(index)) for index in range(5)]
+ )
+
+ assert len(response.confirmations) == 3
+ assert [
+ confirmation.base_offset for confirmation in response.confirmations
+ ] == [0, 2, 4]
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_linger_paces_sequential_direct_sends(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ linger = timedelta(milliseconds=150)
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ mode=DirectProducerConfig(linger_time=linger),
+ )
+ try:
+ await producer.send_one(SendMessage("first"))
+ started_at = time.monotonic()
+ await producer.send_one(SendMessage("second"))
+ elapsed = time.monotonic() - started_at
+
+ assert elapsed >= 0.1
+ finally:
+ await producer.shutdown()
+
+
+class TestBackgroundProducerSends:
+ """Test background acceptance, batching, routing, and graceful flush."""
+
+ @pytest.mark.asyncio
+ async def test_all_send_methods_accept_messages_without_confirmations(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("bg-stream-", min_bytes=20, max_bytes=20)
+ topic_name = unique_name("bg-topic-", min_bytes=20, max_bytes=20)
+ destination_stream = unique_name("to-stream-", min_bytes=20, max_bytes=20)
+ destination_topic = unique_name("to-topic-", min_bytes=20, max_bytes=20)
+ await iggy_client.create_stream(destination_stream)
+ await iggy_client.create_topic(destination_stream, destination_topic, 1)
+
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ partitioning=Partitioning.partition_id(1),
+ mode=BackgroundProducerConfig(
+ num_shards=2,
+ linger_time=timedelta(seconds=60),
+ batch_size=0,
+ batch_length=0,
+ max_in_flight=2,
+ sharding=ProducerSharding.BALANCED,
+ ),
+ topic_partitions_count=3,
+ )
+ try:
+ responses = [
+ await producer.send(
+ [SendMessage("batch-one"), SendMessage("batch-two")]
+ ),
+ await producer.send_one(SendMessage("single")),
+ await producer.send_with_partitioning(
+ [SendMessage("override")], Partitioning.partition_id(2)
+ ),
+ await producer.send_to(
+ destination_stream,
+ destination_topic,
+ [SendMessage("send-to")],
+ Partitioning.partition_id(0),
+ ),
+ ]
+
+ assert all(
+ isinstance(response, SendMessagesResponse) for response in responses
+ )
+ assert all(response.confirmations == [] for response in responses)
+ finally:
+ await producer.shutdown()
+
+ payloads = await poll_payloads(
+ iggy_client, stream_name, topic_name, partition_id=1
+ )
+ assert sorted(payloads) == ["batch-one", "batch-two", "single"]
+ assert await poll_payloads(
+ iggy_client, stream_name, topic_name, partition_id=2
+ ) == ["override"]
+ assert await poll_payloads(
+ iggy_client, destination_stream, destination_topic
+ ) == ["send-to"]
+
+ @pytest.mark.asyncio
+ async def test_batch_length_flushes_after_the_configured_number_of_sends(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("length-", min_bytes=12, max_bytes=12)
+ topic_name = unique_name("topic-", min_bytes=12, max_bytes=12)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(seconds=5),
+ batch_size=0,
+ batch_length=2,
+ ),
+ )
+ try:
+ await producer.send([SendMessage("first-a"), SendMessage("first-b")])
+ await asyncio.sleep(0.05)
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == []
+
+ await producer.send_one(SendMessage("second-send"))
+ assert await wait_for_payloads(
+ iggy_client,
+ stream_name,
+ topic_name,
+ ["first-a", "first-b", "second-send"],
+ ) == ["first-a", "first-b", "second-send"]
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_batch_size_flushes_after_buffered_bytes_reach_the_threshold(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("size-", min_bytes=12, max_bytes=12)
+ topic_name = unique_name("topic-", min_bytes=12, max_bytes=12)
+ first = "a" * 50
+ second = "b" * 50
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(seconds=5),
+ batch_size=200,
+ batch_length=0,
+ ),
+ )
+ try:
+ # With two 12-byte identifiers and the 64-byte message header,
+ # either send is below 200 reported bytes while both exceed it.
+ await producer.send_one(SendMessage(first))
+ await asyncio.sleep(0.05)
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == []
+
+ await producer.send_one(SendMessage(second))
+ assert await wait_for_payloads(
+ iggy_client, stream_name, topic_name, [first, second]
+ ) == [first, second]
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_linger_time_flushes_a_non_empty_buffer(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("linger-", min_bytes=12, max_bytes=12)
+ topic_name = unique_name("topic-", min_bytes=12, max_bytes=12)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(milliseconds=200),
+ batch_size=0,
+ batch_length=0,
+ ),
+ )
+ try:
+ await producer.send_one(SendMessage("linger"))
+ await asyncio.sleep(0.05)
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == []
+ assert await wait_for_payloads(
+ iggy_client, stream_name, topic_name, ["linger"]
+ ) == ["linger"]
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_zero_sentinels_use_one_shard_and_unlimited_capacity(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("zero-", min_bytes=12, max_bytes=12)
+ topic_name = unique_name("topic-", min_bytes=12, max_bytes=12)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ num_shards=0,
+ linger_time=timedelta(0),
+ batch_size=0,
+ batch_length=0,
+ max_buffer_size=0,
+ max_in_flight=0,
+ ),
+ )
+ try:
+ response = await producer.send_one(SendMessage("zero sentinels"))
+ assert response.confirmations == []
+ finally:
+ await producer.shutdown()
+
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == [
+ "zero sentinels"
+ ]
+
+ @pytest.mark.asyncio
+ async def test_ordered_sharding_preserves_order_for_one_destination(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("ordered-", min_bytes=16, max_bytes=16)
+ topic_name = unique_name("topic-", min_bytes=16, max_bytes=16)
+ expected = [str(index) for index in range(30)]
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ num_shards=4,
+ linger_time=timedelta(0),
+ batch_size=0,
+ batch_length=1,
+ max_in_flight=4,
+ sharding=ProducerSharding.ORDERED,
+ ),
+ )
+ try:
+ for payload in expected:
+ await producer.send_one(SendMessage(payload))
+ finally:
+ await producer.shutdown()
+
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == expected
+
+
+class TestBackgroundProducerBackpressure:
+ """Test each byte-budget backpressure policy through the Python API."""
+
+ @staticmethod
+ def config(
+ failure_mode: BackpressureMode,
+ *,
+ linger_time: timedelta,
+ ) -> BackgroundProducerConfig:
+ return BackgroundProducerConfig(
+ linger_time=linger_time,
+ batch_size=0,
+ batch_length=0,
+ max_buffer_size=250,
+ failure_mode=failure_mode,
+ )
+
+ @pytest.mark.asyncio
+ async def test_fail_immediately_rejects_a_send_when_the_buffer_is_full(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name("fail-", min_bytes=12, max_bytes=12),
+ unique_name("topic-", min_bytes=12, max_bytes=12),
+ mode=self.config(
+ BackpressureMode.fail_immediately(),
+ linger_time=timedelta(seconds=5),
+ ),
+ )
+ try:
+ assert (await producer.send_one(SendMessage("a" * 128))).confirmations == []
+ with pytest.raises(RuntimeError, match="(?i)buffer|overflow"):
+ await producer.send_one(SendMessage("b" * 128))
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_block_with_timeout_waits_then_reports_a_timeout(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name("timeout-", min_bytes=16, max_bytes=16),
+ unique_name("topic-", min_bytes=16, max_bytes=16),
+ mode=self.config(
+ BackpressureMode.block_with_timeout(timedelta(milliseconds=100)),
+ linger_time=timedelta(seconds=5),
+ ),
+ )
+ try:
+ await producer.send_one(SendMessage("a" * 128))
+ started_at = time.monotonic()
+ with pytest.raises(RuntimeError, match="(?i)timeout"):
+ await producer.send_one(SendMessage("b" * 128))
+ elapsed = time.monotonic() - started_at
+
+ assert 0.05 <= elapsed < 1
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_block_waits_until_a_previous_batch_releases_capacity(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name("block-", min_bytes=12, max_bytes=12),
+ unique_name("topic-", min_bytes=12, max_bytes=12),
+ mode=self.config(
+ BackpressureMode.block(),
+ linger_time=timedelta(milliseconds=200),
+ ),
+ )
+ try:
+ await producer.send_one(SendMessage("a" * 128))
+ started_at = time.monotonic()
+ response = await producer.send_one(SendMessage("b" * 128))
+ elapsed = time.monotonic() - started_at
+
+ assert response.confirmations == []
+ assert elapsed >= 0.1
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_batch_larger_than_the_total_budget_fails_without_blocking(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name("oversize-", min_bytes=16, max_bytes=16),
+ unique_name("topic-", min_bytes=16, max_bytes=16),
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(seconds=5),
+ batch_size=0,
+ batch_length=0,
+ max_buffer_size=100,
+ failure_mode=BackpressureMode.block(),
+ ),
+ )
+ try:
+ with pytest.raises(RuntimeError, match="(?i)buffer|overflow"):
+ await asyncio.wait_for(
+ producer.send_one(SendMessage("larger than the budget")),
+ timeout=1,
+ )
+ finally:
+ await producer.shutdown()
+
+
+class TestProducerLifecycle:
+ """Test producer ownership, shutdown, and asynchronous context management."""
+
+ @pytest.mark.asyncio
+ async def test_linger_sends_run_concurrently_under_shared_lifecycle_access(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ linger_seconds = 2
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ mode=DirectProducerConfig(linger_time=timedelta(seconds=linger_seconds)),
+ )
+ try:
+ await producer.send_one(SendMessage("prime linger"))
+ started_at = time.monotonic()
+ responses = await asyncio.gather(
+ producer.send_one(SendMessage("concurrent one")),
+ producer.send_one(SendMessage("concurrent two")),
+ )
+ elapsed = time.monotonic() - started_at
+
+ assert len(responses) == 2
+ assert all(len(response.confirmations) == 1 for response in responses)
+ # Both reads wait against the same timestamp. A mutex around the
+ # producer would make the second wait through another full linger.
+ assert 1.2 <= elapsed < 3.6
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_shutdown_is_sequentially_and_concurrently_idempotent(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+
+ assert await asyncio.gather(producer.shutdown(), producer.shutdown()) == [
+ None,
+ None,
+ ]
+ assert await producer.shutdown() is None
+
+ @pytest.mark.asyncio
+ async def test_background_shutdown_is_idempotent_and_flushes_accepted_messages(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("shutdown-", min_bytes=20, max_bytes=20)
+ topic_name = unique_name("topic-", min_bytes=20, max_bytes=20)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(seconds=60),
+ batch_size=0,
+ batch_length=0,
+ ),
+ )
+ await producer.send([SendMessage("one"), SendMessage("two")])
+
+ assert await asyncio.gather(producer.shutdown(), producer.shutdown()) == [
+ None,
+ None,
+ ]
+ assert await producer.shutdown() is None
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == [
+ "one",
+ "two",
+ ]
+
+ @pytest.mark.asyncio
+ async def test_every_send_rejects_use_after_shutdown(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ producer = await iggy_client.producer(stream_name, topic_name)
+ await producer.shutdown()
+
+ calls = [
+ producer.send([]),
+ producer.send_one(SendMessage("closed")),
+ producer.send_with_partitioning([], None),
+ producer.send_to(stream_name, topic_name, [], None),
+ ]
+ for call in calls:
+ with pytest.raises(RuntimeError, match="closed|shut down"):
+ await call
+
+ @pytest.mark.asyncio
+ async def test_shutdown_waits_for_a_send_holding_lifecycle_access(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ mode=DirectProducerConfig(linger_time=timedelta(milliseconds=150)),
+ )
+ await producer.send_one(SendMessage("prime linger"))
+
+ sending = asyncio.ensure_future(producer.send_one(SendMessage("in flight")))
+
+ async def wait_for_active_send():
+ # pyrefly: ignore # missing-attribute
+ while not producer._is_send_active():
+ await asyncio.sleep(0)
+
+ await asyncio.wait_for(wait_for_active_send(), timeout=1)
+ shutting_down = asyncio.ensure_future(producer.shutdown())
+
+ response, shutdown_result = await asyncio.wait_for(
+ asyncio.gather(sending, shutting_down), timeout=2
+ )
+ assert len(response.confirmations) == 1
+ assert shutdown_result is None
+ with pytest.raises(RuntimeError, match="closed|shut down"):
+ await producer.send_one(SendMessage("too late"))
+
+ @pytest.mark.asyncio
+ async def test_background_send_racing_shutdown_completes_without_deadlock(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name("race-", min_bytes=12, max_bytes=12)
+ topic_name = unique_name("topic-", min_bytes=12, max_bytes=12)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(milliseconds=200),
+ batch_size=0,
+ batch_length=0,
+ max_buffer_size=250,
+ failure_mode=BackpressureMode.block(),
+ ),
+ )
+ await producer.send_one(SendMessage("a" * 128))
+ sending = asyncio.ensure_future(producer.send_one(SendMessage("b" * 128)))
+
+ async def wait_for_blocked_send():
+ # pyrefly: ignore # missing-attribute
+ while not producer._is_send_active():
+ await asyncio.sleep(0)
+
+ await asyncio.wait_for(wait_for_blocked_send(), timeout=1)
+ shutting_down = asyncio.ensure_future(producer.shutdown())
+ response, shutdown_result = await asyncio.wait_for(
+ asyncio.gather(sending, shutting_down), timeout=2
+ )
+
+ assert response.confirmations == []
+ assert shutdown_result is None
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == [
+ "a" * 128,
+ "b" * 128,
+ ]
+
+ @pytest.mark.asyncio
+ async def test_async_context_manager_returns_self_and_closes_normally(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+
+ async with producer as entered:
+ assert entered is producer
+ assert (
+ len((await entered.send_one(SendMessage("inside"))).confirmations) == 1
+ )
+
+ with pytest.raises(RuntimeError, match="closed|shut down"):
+ await producer.send_one(SendMessage("outside"))
+
+ @pytest.mark.asyncio
+ async def test_async_context_manager_closes_on_exception_without_suppressing_it(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+
+ with pytest.raises(LookupError, match="application failure"):
+ async with producer:
+ raise LookupError("application failure")
+
+ with pytest.raises(RuntimeError, match="closed|shut down"):
+ await producer.send_one(SendMessage("outside"))
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "raises_inside", [False, True], ids=["normal", "exception"]
+ )
+ async def test_background_context_manager_flushes_on_every_exit(
+ self, iggy_client: IggyClient, unique_name, raises_inside: bool
+ ):
+ stream_name = unique_name("context-", min_bytes=20, max_bytes=20)
+ topic_name = unique_name("topic-", min_bytes=20, max_bytes=20)
+ producer = await iggy_client.producer(
+ stream_name,
+ topic_name,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(seconds=60),
+ batch_size=0,
+ batch_length=0,
+ ),
+ )
+
+ async def use_context():
+ async with producer:
+ await producer.send_one(SendMessage("buffered"))
+ if raises_inside:
+ raise LookupError("application failure")
+
+ if raises_inside:
+ with pytest.raises(LookupError, match="application failure"):
+ await use_context()
+ else:
+ await use_context()
+
+ assert await poll_payloads(iggy_client, stream_name, topic_name) == ["buffered"]
+ with pytest.raises(RuntimeError, match="closed|shut down"):
+ await producer.send_one(SendMessage("outside"))
+
+
+class TestProducerValidationAndRetries:
+ """Test stable error categories and observable retry policy behavior."""
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("kwargs", "expected_exception"),
+ [
+ ({"partitioning": 0}, TypeError),
+ ({"mode": object()}, TypeError),
+ ({"topic_partitions_count": -1}, ValueError),
+ ({"topic_partitions_count": 2**32}, ValueError),
+ ({"topic_partitions_count": 2**63}, OverflowError),
+ ({"topic_message_expiry": timedelta(seconds=1)}, TypeError),
+ ({"topic_max_size": 1000}, TypeError),
+ ({"send_retries": -1}, ValueError),
+ ({"send_retries": 2**32}, ValueError),
+ ({"send_retries": 2**63}, OverflowError),
+ ({"send_retry_interval": timedelta(0)}, ValueError),
+ ({"send_retry_interval": timedelta(microseconds=-1)}, ValueError),
+ ({"send_retry_interval": 1}, TypeError),
+ ],
+ )
+ async def test_producer_configuration_errors_have_stable_categories(
+ self,
+ iggy_client: IggyClient,
+ unique_name,
+ kwargs: dict,
+ expected_exception: type[Exception],
+ ):
+ with pytest.raises(expected_exception):
+ await iggy_client.producer(unique_name(), unique_name(), **kwargs)
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "mode",
+ [
+ BackgroundProducerConfig(max_buffer_size=2**64 - 1),
+ BackgroundProducerConfig(max_in_flight=sys.maxsize),
+ ],
+ ids=["max-buffer-size", "max-in-flight"],
+ )
+ async def test_background_limits_that_would_panic_rust_are_value_errors(
+ self,
+ iggy_client: IggyClient,
+ unique_name,
+ mode: BackgroundProducerConfig,
+ ):
+ with pytest.raises(ValueError, match="must not exceed"):
+ await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ mode=mode,
+ )
+
+ @pytest.mark.asyncio
+ async def test_bound_destination_requires_names(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await iggy_client.producer(1, unique_name())
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await iggy_client.producer(unique_name(), 1)
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("stream", "topic"),
+ [
+ ("", "topic"),
+ ("stream", ""),
+ ("x" * 256, "topic"),
+ ("stream", "x" * 256),
+ ],
+ )
+ async def test_bound_destination_rejects_invalid_names_as_value_errors(
+ self, iggy_client: IggyClient, stream: str, topic: str
+ ):
+ with pytest.raises(ValueError):
+ await iggy_client.producer(stream, topic)
+
+ @pytest.mark.asyncio
+ async def test_send_arguments_reject_low_level_shorthands_and_wrong_types(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+ try:
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await producer.send((SendMessage("tuple"),))
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await producer.send(["payload"])
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await producer.send_one("payload")
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await producer.send_with_partitioning([SendMessage("integer")], 0)
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-type
+ await producer.send_to(object(), "topic", [SendMessage("bad stream")])
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("identifier", [-1, 2**32, 2**63])
+ async def test_send_to_identifier_overflow_is_not_folded_into_type_error(
+ self, iggy_client: IggyClient, unique_name, identifier: int
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+ try:
+ for stream, topic in [(identifier, "topic"), ("stream", identifier)]:
+ with pytest.raises(OverflowError):
+ await producer.send_to(
+ stream,
+ topic,
+ [SendMessage("outside u32")],
+ )
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_server_send_errors_preserve_recovery_state(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ send_retries=0,
+ )
+ try:
+ with pytest.raises(ProducerSendError) as raised:
+ await producer.send_with_partitioning(
+ [SendMessage("missing partition")],
+ Partitioning.partition_id(1),
+ )
+
+ error = raised.value
+ assert isinstance(error, RuntimeError)
+ assert error.cause
+ assert isinstance(error.__cause__, RuntimeError)
+ assert str(error.__cause__) == error.cause
+ assert len(error.failed) == 1
+ assert isinstance(error.failed[0], SendMessage)
+ assert error.committed == []
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("send_retries", [None, 0])
+ async def test_none_and_zero_disable_retries(
+ self, iggy_client: IggyClient, unique_name, send_retries: int | None
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ send_retries=send_retries,
+ send_retry_interval=timedelta(seconds=5),
+ )
+ try:
+ with pytest.raises(RuntimeError):
+ await asyncio.wait_for(
+ producer.send_to(
+ unique_name(), unique_name(), [SendMessage("single attempt")]
+ ),
+ timeout=1,
+ )
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_explicit_none_interval_retries_without_delay(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ send_retries=10,
+ send_retry_interval=None,
+ )
+ try:
+ with pytest.raises(RuntimeError):
+ await asyncio.wait_for(
+ producer.send_to(
+ unique_name(), unique_name(), [SendMessage("immediate retries")]
+ ),
+ timeout=1,
+ )
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_custom_retry_interval_paces_later_retries(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(
+ unique_name(),
+ unique_name(),
+ send_retries=2,
+ send_retry_interval=timedelta(milliseconds=150),
+ )
+ try:
+ started_at = time.monotonic()
+ with pytest.raises(RuntimeError):
+ await producer.send_to(
+ unique_name(), unique_name(), [SendMessage("paced retries")]
+ )
+ elapsed = time.monotonic() - started_at
+
+ assert elapsed >= 0.1
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_default_retry_policy_retains_the_rust_interval(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ producer = await iggy_client.producer(unique_name(), unique_name())
+ try:
+ started_at = time.monotonic()
+ with pytest.raises(RuntimeError):
+ await asyncio.wait_for(
+ producer.send_to(
+ unique_name(), unique_name(), [SendMessage("default retries")]
+ ),
+ timeout=6,
+ )
+ elapsed = time.monotonic() - started_at
+
+ # The first retry is immediate. Three retries with the one-second
+ # default wait for two later interval ticks.
+ assert elapsed >= 1.5
+ finally:
+ await producer.shutdown()
+
+ @pytest.mark.asyncio
+ async def test_background_write_retries_until_a_destination_appears(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ destination_stream = unique_name("retry-stream-", min_bytes=24, max_bytes=24)
+ destination_topic = unique_name("retry-topic-", min_bytes=24, max_bytes=24)
+ producer = await iggy_client.producer(
+ unique_name("bound-stream-", min_bytes=24, max_bytes=24),
+ unique_name("bound-topic-", min_bytes=24, max_bytes=24),
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(0),
+ batch_length=1,
+ ),
+ send_retries=10,
+ send_retry_interval=timedelta(milliseconds=100),
+ )
+ try:
+ response = await producer.send_to(
+ destination_stream,
+ destination_topic,
+ [SendMessage("eventual destination")],
+ )
+ assert response.confirmations == []
+
+ # The first retry is immediate. Create the destination before a
+ # later interval tick so the worker can recover asynchronously.
+ await asyncio.sleep(0.03)
+ await iggy_client.create_stream(destination_stream)
+ await iggy_client.create_topic(destination_stream, destination_topic, 1)
+ finally:
+ await producer.shutdown()
+
+ assert await wait_for_payloads(
+ iggy_client,
+ destination_stream,
+ destination_topic,
+ ["eventual destination"],
+ ) == ["eventual destination"]
+
+ @pytest.mark.asyncio
+ async def test_background_write_reconnects_after_server_restart(
+ self,
+ tmp_path: Path,
+ unique_name,
+ ):
+ data_path = tmp_path / "restartable-server"
+ server = spawn_restartable_server(data_path, "127.0.0.1:0")
+ producer: IggyProducer | None = None
+
+ try:
+ host, port = await discover_server_address(data_path, server)
+ await asyncio.to_thread(wait_for_server, host, port, 15, 1)
+ client = IggyClient.from_connection_string(
+ f"iggy+tcp://iggy:iggy@{host}:{port}"
+ "?reconnection_interval=100ms&reestablish_after=0"
+ )
+ await client.connect()
+
+ stream = unique_name("reconnect-stream-", min_bytes=28, max_bytes=28)
+ topic = unique_name("reconnect-topic-", min_bytes=28, max_bytes=28)
+ producer = await client.producer(
+ stream,
+ topic,
+ mode=BackgroundProducerConfig(
+ linger_time=timedelta(0),
+ batch_length=1,
+ ),
+ send_retries=20,
+ send_retry_interval=timedelta(milliseconds=100),
+ )
+
+ await stop_restartable_server(server)
+ server = None
+ accepted = await producer.send_one(SendMessage("after restart"))
+ assert accepted.confirmations == []
+
+ server = spawn_restartable_server(data_path, f"{host}:{port}")
+ await asyncio.to_thread(wait_for_server, host, port, 15, 1)
+ await asyncio.wait_for(producer.shutdown(), timeout=15)
+ producer = None
+
+ assert await wait_for_payloads(
+ client,
+ stream,
+ topic,
+ ["after restart"],
+ timeout=10,
+ ) == ["after restart"]
+ finally:
+ if producer is not None:
+ if server is None:
+ server = spawn_restartable_server(data_path, f"{host}:{port}")
+ await asyncio.to_thread(wait_for_server, host, port, 15, 1)
+ with contextlib.suppress(RuntimeError, TimeoutError):
+ await asyncio.wait_for(producer.shutdown(), timeout=5)
+ if server is not None:
+ await stop_restartable_server(server)