jiengup opened a new pull request, #4156:
URL: https://github.com/apache/iggy/pull/4156
## Which issue does this PR address?
Closes #4019
## Rationale
The Python SDK needs a stable, reusable producer abstraction that follows
the Rust core for direct server-confirmed sends and buffered background
dispatch while preserving the existing low-level API.
## What changed?
Python callers previously had only `IggyClient.send_messages()`, so each
send had to carry its destination and none of the Rust high-level producer
lifecycle was exposed. Async `IggyClient.producer()` now builds and initializes
an `IggyProducer` with direct or background configuration, default and per-call
partitioning, optional stream and topic creation, retry settings, four send
methods, partial-send recovery through `ProducerSendError`, and idempotent
async shutdown and context-manager support. Background mode exposes batching,
ordered or balanced sharding, bounded buffers and in-flight requests, all three
backpressure policies, and flush-on-shutdown semantics. The generated stub,
READMEs, runnable direct and background producer examples, paired consumer
example, and Rust and Python tests cover the public surface while
`IggyClient.send_messages()` remains unchanged.
Python SDK high-level consumer and producer API examples have also been
added in this PR.
### Rust high-level API parity
- `IggyClient.producer()` binds a stream and topic, builds and initializes
the producer before returning, uses balanced partitioning by default, and
optionally creates missing resources with the requested partition count,
expiry, and maximum size.
- `IggyProducer` exposes `send()`, `send_one()`, `send_with_partitioning()`,
and `send_to()`. Per-call partitioning overrides fall back to the producer
default, while `send_to()` targets another existing destination without
creating it.
- Direct mode waits for server confirmations, applies batch-length and
linger settings, and retains the Rust at-least-once retry semantics. The first
retry is immediate, later retries use the configured interval, and a lost
response can result in duplicate delivery.
- Background mode completes when the dispatcher accepts a send and therefore
returns an empty confirmation list. It maps the Rust shard count, linger, byte
and queued-send batching thresholds, global buffer limit, in-flight limit,
ordered or balanced sharding, and block, timed-block, or fail-immediately
backpressure modes.
- Ordered sharding preserves dispatch order for one destination, while
balanced sharding distributes consecutive sends across workers without
guaranteeing order. Shutdown drains the queues and flushes all accepted
background messages.
- Empty batches, initialization failures, retry and reconnect behavior, and
partial direct-send recovery follow the Rust implementation. The existing
low-level `IggyClient.send_messages()` API remains available.
### Python-specific behavior
- Producer creation, sending, and shutdown are awaitable. `mode=None`
selects direct mode, durations use `datetime.timedelta`, configuration objects
are immutable and keyword-only, and sharding strategies use uppercase Python
enum values.
- Rust's consuming `shutdown(self)` is adapted through shared lifecycle
state. Shutdown is idempotent, waits for active sends, remains safe when called
concurrently, and every send after shutdown raises `RuntimeError`.
- `IggyProducer` implements the asynchronous context-manager protocol,
including deterministic shutdown on normal and exceptional exits. Cleanup does
not depend on `__del__`.
- Rust partial-send failures are exposed as `ProducerSendError`, a
`RuntimeError` subclass containing the underlying `cause`, confirmations for
`committed` chunks, and the remaining `failed` messages.
- Invalid Python types, durations, identifiers, and numeric ranges produce
stable `TypeError`, `ValueError`, or `OverflowError` categories. The generated
`.pyi` stub exposes valid Python defaults without leaking internal binding
types.
- Python callbacks for custom sharding and background errors are
intentionally not exposed. Background writes that still fail after retries
continue to be logged by the Rust worker, and their unconfirmed messages are
dropped.
- README documentation and runnable direct producer, background producer,
and paired consumer examples describe these semantics, with Rust and Python
tests covering configuration, initialization, routing, batching, retries,
reconnects, backpressure, concurrency, recovery state, and shutdown.
## Local Execution
- Passed:
- `cargo fmt --all -- --check`
- `cargo clippy --all-features --all-targets -- -D warnings`
- `cargo test -p iggy --lib clients::producer::tests` (2 passed)
- `cargo fmt --manifest-path foreign/python/Cargo.toml -- --check`
- `cargo clippy --manifest-path foreign/python/Cargo.toml --all-targets
--all-features -- -D warnings`
- `cargo test --manifest-path foreign/python/Cargo.toml` (20 passed)
- `cd foreign/python && uv run --no-sync ruff check .
../../examples/python`
- `cd foreign/python && uv run --no-sync ruff format --check .
../../examples/python`
- `cd foreign/python && uv run --no-sync pyrefly check .`
- `cd foreign/python && uv run --no-sync pytest tests/test_producer.py`
(100 passed)
- `./scripts/ci/markdownlint.sh --check foreign/python/README.md
examples/python/README.md`
- `git diff --check upstream/master...HEAD`
- Pre-commit and pre-push hooks ran.
## AI Usage
OpenAI Codex.
Used to inspect the branch diff against `upstream/master`, cross-check the
implementation against #4019, rerun targeted validation, and draft this PR
description. No implementation files were modified while preparing the
description. The described behavior was verified against the current diff and
the local results above.
Yes, I can explain every submitted change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
