You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "jiengup (via GitHub)" <gi...@apache.org> on 2026/08/20 06:53:21 UTC
[PR] Python SDK: support message partitioning strategies [iggy]
jiengup opened a new pull request, #3927:
URL: https://github.com/apache/iggy/pull/3927
## Which issue does this PR address?
Closes #3896
## Rationale
The Python SDK only supported sending messages to an explicit partition, while the Rust SDK also supports balanced and message-key partitioning.
## What changed?
`IggyClient.send_messages()` now accepts a Python `Partitioning` object supporting fixed, balanced, and message-key routing. Existing integer partition IDs remain fully compatible, while string message keys are encoded as UTF-8 and validated against the 1–255 byte limit.
Type stubs, tests, and a runnable Python example covering all three strategies are included.
## Local Execution
- Passed all 87 tests in `foreign/python/tests/test_message_operations.py`
- Passed Rust formatting and Clippy checks
- Passed Ruff and Pyrefly checks for the Python SDK and examples
- Successfully ran the partitioning example against a local Iggy server
- Pre-commit hooks ran and passed
## AI Usage
1. Tool: OpenAI Codex
2. Scope: Repository analysis, PyO3 implementation, tests, type stubs, documentation, and the Python example
3. Verification: Reviewed the generated changes, ran the complete message operations test suite, static checks, pre-commit hooks, and the example against a local server
4. Yes, I can explain every line of the changes
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz merged PR #3927:
URL: https://github.com/apache/iggy/pull/3927
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3839566051
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -109,6 +150,87 @@ async def test_send_messages_reports_committed_confirmation(
assert confirmation.partition_id == partition_id
assert confirmation.base_offset == polled_messages[0].offset()
+ @pytest.mark.asyncio
+ async def test_send_messages_with_partition_id_strategy(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 2
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=3
+ )
+
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=Partitioning.partition_id(partition_id),
+ messages=[Message("fixed partition")],
+ )
+
+ assert len(response.confirmations) == 1
+ assert response.confirmations[0].partition_id == partition_id
+
+ @pytest.mark.asyncio
+ async def test_send_messages_with_balanced_strategy(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partitions_count = 3
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=partitions_count,
+ )
+
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=Partitioning.balanced(),
+ messages=[Message("balanced")],
+ )
+
+ assert len(response.confirmations) == 1
+ assert response.confirmations[0].partition_id < partitions_count
Review Comment:
+1
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Python SDK: support message partitioning strategies [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3927:
URL: https://github.com/apache/iggy/pull/3927#issuecomment-5352464370
<!-- iggy-pr-triage-welcome -->
Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
Slash commands (own line, regular comment) move it around the queue:
- `/ready` - back to `S-waiting-on-review` after addressing feedback
- `/author` - flip to `S-waiting-on-author` while you finish changes
- `/request-review @user-or-team` - request a reviewer
See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "jiengup (via GitHub)" <gi...@apache.org>.
jiengup commented on PR #3927:
URL: https://github.com/apache/iggy/pull/3927#issuecomment-5442151165
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "jiengup (via GitHub)" <gi...@apache.org>.
jiengup commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3894971454
##########
examples/python/basic/producer.py:
##########
@@ -106,7 +106,11 @@ async def produce_messages(client: IggyClient):
await client.send_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
- partitioning=PARTITION_ID,
+ # A fixed strategy sends the whole batch to this partition.
+ # For topics with multiple partitions, Partitioning.balanced()
+ # distributes batches round-robin, while
+ # Partitioning.messages_key(key) keeps equal keys on the same partition.
+ partitioning=Partitioning.partition_id(PARTITION_ID),
Review Comment:
I just removed the criteria in the issue to keep the basic example simple to understand.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3839458246
##########
examples/python/README.md:
##########
Review Comment:
Follows from above comment -- not required.
##########
foreign/python/src/partitioning.rs:
##########
@@ -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.
+
+use iggy::prelude::Partitioning as RustPartitioning;
+use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
+use pyo3_stub_gen::{
+ derive::{gen_stub_pyclass, gen_stub_pymethods},
+ impl_stub_type,
+};
+
+/// Defines how a batch of messages is assigned to a topic partition.
+#[derive(Clone)]
+#[pyclass(from_py_object)]
+#[gen_stub_pyclass]
+pub struct Partitioning {
+ pub(crate) inner: RustPartitioning,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Partitioning {
+ /// Routes the batch to partitions using server-side round-robin selection.
+ #[staticmethod]
+ pub fn balanced() -> Self {
+ Self {
+ inner: RustPartitioning::balanced(),
+ }
+ }
+
+ /// Routes the batch to the specified partition.
+ #[staticmethod]
+ pub fn partition_id(partition_id: u32) -> Self {
+ Self {
+ inner: RustPartitioning::partition_id(partition_id),
+ }
+ }
+
+ /// Routes the batch using a binary key hashed by the server.
+ ///
+ /// String keys are encoded as UTF-8. The encoded key must contain between
+ /// 1 and 255 bytes.
Review Comment:
Change to
```rust
/// Routes the batch to one partition selected by hashing `key`.
///
/// `key` may be `str` or `bytes`. Strings are encoded as UTF-8; the encoded
/// key must contain between 1 and 255 bytes.
///
/// Raises:
/// ValueError: If the encoded key is empty or exceeds 255 bytes.
/// TypeError: If `key` is not `str` or `bytes`.
```
##########
examples/python/partitioning/producer.py:
##########
Review Comment:
I'd rather a new example like this not be created, unless it is a capability that the rust sdk does not offer. Better to add comments to an existing example writing how can message partitioning options will result in different outcomes, if any.
##########
foreign/python/src/partitioning.rs:
##########
@@ -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.
+
+use iggy::prelude::Partitioning as RustPartitioning;
+use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
+use pyo3_stub_gen::{
+ derive::{gen_stub_pyclass, gen_stub_pymethods},
+ impl_stub_type,
+};
+
+/// Defines how a batch of messages is assigned to a topic partition.
+#[derive(Clone)]
+#[pyclass(from_py_object)]
+#[gen_stub_pyclass]
+pub struct Partitioning {
+ pub(crate) inner: RustPartitioning,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Partitioning {
+ /// Routes the batch to partitions using server-side round-robin selection.
+ #[staticmethod]
+ pub fn balanced() -> Self {
+ Self {
+ inner: RustPartitioning::balanced(),
+ }
+ }
+
+ /// Routes the batch to the specified partition.
Review Comment:
Change to
```rust
/// Routes the batch to the specified partition.
///
/// `partition_id` must be between 0 and `2**32 - 1`. The topic must contain
/// that partition when the batch is sent.
```
##########
foreign/python/src/client.rs:
##########
@@ -1023,7 +1029,7 @@ impl IggyClient {
let stream = Identifier::try_from(stream)?;
let topic = Identifier::try_from(topic)?;
- let partitioning = Partitioning::partition_id(partitioning);
+ let partitioning = partitioning.into();
Review Comment:
Let's change this into `from::`. I like to be explicit about what is being converted into what in code. `into` does not help a first-time reader understand this without searching the code base.
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -24,12 +24,53 @@
HeaderKey,
HeaderValue,
IggyClient,
+ Partitioning,
PollingStrategy,
UserHeaders,
)
from apache_iggy import SendMessage as Message
+class TestPartitioning:
+ """Test message partitioning strategy construction."""
+
+ @pytest.mark.unit
+ def test_balanced_and_partition_id_strategies(self):
+ assert isinstance(Partitioning.balanced(), Partitioning)
+ assert isinstance(Partitioning.partition_id(1), Partitioning)
Review Comment:
Let's add an assertion for `partition_id(2**32 -1)`
##########
foreign/python/src/partitioning.rs:
##########
@@ -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.
+
+use iggy::prelude::Partitioning as RustPartitioning;
+use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
+use pyo3_stub_gen::{
+ derive::{gen_stub_pyclass, gen_stub_pymethods},
+ impl_stub_type,
+};
+
+/// Defines how a batch of messages is assigned to a topic partition.
+#[derive(Clone)]
+#[pyclass(from_py_object)]
+#[gen_stub_pyclass]
+pub struct Partitioning {
+ pub(crate) inner: RustPartitioning,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Partitioning {
+ /// Routes the batch to partitions using server-side round-robin selection.
Review Comment:
Change to "Routes the batch to one partition selected by round-robin."
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -109,6 +150,87 @@ async def test_send_messages_reports_committed_confirmation(
assert confirmation.partition_id == partition_id
assert confirmation.base_offset == polled_messages[0].offset()
+ @pytest.mark.asyncio
+ async def test_send_messages_with_partition_id_strategy(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 2
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=3
+ )
+
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=Partitioning.partition_id(partition_id),
+ messages=[Message("fixed partition")],
+ )
+
+ assert len(response.confirmations) == 1
+ assert response.confirmations[0].partition_id == partition_id
+
+ @pytest.mark.asyncio
+ async def test_send_messages_with_balanced_strategy(
Review Comment:
Let's change it like this: replace this test's single send with four sends using `Partitioning.balanced()`, collect the confirmation partition IDs, and assert `[0, 1, 2, 0]`.
##########
foreign/python/src/client.rs:
##########
@@ -1000,13 +1001,18 @@ impl IggyClient {
/// confirmations, or a PyRuntimeError on failure. The confirmation list is
/// empty when the server reports no offsets, and the legacy server never
/// reports any.
+ ///
+ /// `partitioning` is required. Pass `Partitioning.balanced()`,
+ /// `Partitioning.partition_id(id)`, or `Partitioning.messages_key(key)`.
+ /// An integer remains supported as shorthand for `partition_id`.
Review Comment:
Lets change this to
```rust
/// Sends a batch of messages to a topic using the selected partitioning strategy.
///
/// Args:
/// stream: Stream identifier as `str | int`.
/// topic: Topic identifier as `str | int`.
/// partitioning: A `Partitioning` strategy or an integer partition ID.
/// Use `Partitioning.balanced()`, `Partitioning.partition_id(id)`, or
/// `Partitioning.messages_key(key)`. An integer is shorthand for
/// `Partitioning.partition_id(id)`.
/// messages: Messages to send as `list[SendMessage]`.
///
/// Returns:
/// An awaitable that resolves to `SendMessagesResponse`. Its confirmations
/// report the committed partition and batch base offset. The list is empty
/// when the server reports no offsets, including on the legacy server.
///
/// Raises:
/// ValueError: If a string stream or topic identifier is invalid.
/// TypeError: If `partitioning` or `messages` has an unsupported type.
/// OverflowError: If a numeric stream, topic, or partition ID is outside
/// the supported unsigned 32-bit range.
/// RuntimeError: If the request fails.
```
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -24,12 +24,53 @@
HeaderKey,
HeaderValue,
IggyClient,
+ Partitioning,
PollingStrategy,
UserHeaders,
)
from apache_iggy import SendMessage as Message
+class TestPartitioning:
+ """Test message partitioning strategy construction."""
+
+ @pytest.mark.unit
+ def test_balanced_and_partition_id_strategies(self):
+ assert isinstance(Partitioning.balanced(), Partitioning)
+ assert isinstance(Partitioning.partition_id(1), Partitioning)
+
+ @pytest.mark.unit
+ @pytest.mark.parametrize("key", [b"customer-42", "customer-42", "客户-42"])
+ def test_messages_key_accepts_bytes_and_strings(self, key):
+ assert isinstance(Partitioning.messages_key(key), Partitioning)
+
+ @pytest.mark.unit
+ @pytest.mark.parametrize("key", [b"a" * 255, "a" * 255])
+ def test_messages_key_accepts_255_bytes(self, key):
Review Comment:
Let's add `"界" * 85` as a case for this as well.
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -109,6 +150,87 @@ async def test_send_messages_reports_committed_confirmation(
assert confirmation.partition_id == partition_id
assert confirmation.base_offset == polled_messages[0].offset()
+ @pytest.mark.asyncio
+ async def test_send_messages_with_partition_id_strategy(
Review Comment:
In this test, let's also send one message with `Partitioning.partition_id(3)` and assert `RuntimeError`
##########
foreign/python/tests/test_message_operations.py:
##########
Review Comment:
Something that did not land anywhere: In `TestMessageOperations.test_send_messages_with_partition_id_strategy`, send two messages instead of one and retain the existing single-confirmation assertion.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3927:
URL: https://github.com/apache/iggy/pull/3927#issuecomment-5352556181
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3927?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 83.76%. Comparing base ([`3cd0860`](https://app.codecov.io/gh/apache/iggy/commit/3cd0860a97492cc566d3c2e3b1752c18168b7580?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f715f3c`](https://app.codecov.io/gh/apache/iggy/commit/f715f3cf758f4d5ea373c567dbe731a63524dc4c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3927 +/- ##
=========================================
Coverage 83.76% 83.76%
Complexity 1358 1358
=========================================
Files 1212 1213 +1
Lines 166189 166215 +26
Branches 133663 133663
=========================================
+ Hits 139210 139236 +26
Misses 23342 23342
Partials 3637 3637
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.55% <ø> (ø)` | |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.67% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.08% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.11% <100.00%> (+0.11%)` | :arrow_up: |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.48% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `95.84% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3927/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.32% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3927?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [foreign/python/src/client.rs](https://app.codecov.io/gh/apache/iggy/pull/3927?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fclient.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2NsaWVudC5ycw==) | `99.85% <100.00%> (+<0.01%)` | :arrow_up: |
| [foreign/python/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/3927?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2xpYi5ycw==) | `100.00% <100.00%> (ø)` | |
| [foreign/python/src/partitioning.rs](https://app.codecov.io/gh/apache/iggy/pull/3927?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fpartitioning.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL3BhcnRpdGlvbmluZy5ycw==) | `100.00% <100.00%> (ø)` | |
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "jiengup (via GitHub)" <gi...@apache.org>.
jiengup commented on PR #3927:
URL: https://github.com/apache/iggy/pull/3927#issuecomment-5439438438
[apache_iggy.pyi](https://github.com/apache/iggy/pull/3927/changes#diff-df90a0f5aa5c851a60a7fd6dcaadde2b72d9c8cc8f8d3d695f9d86865b283369) is re-generated through `stub_gen` binary and formatted by `ruff`, **some changes are unrelated to this PR.**
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3839551994
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -109,6 +150,87 @@ async def test_send_messages_reports_committed_confirmation(
assert confirmation.partition_id == partition_id
assert confirmation.base_offset == polled_messages[0].offset()
+ @pytest.mark.asyncio
+ async def test_send_messages_with_partition_id_strategy(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 2
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=3
+ )
+
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=Partitioning.partition_id(partition_id),
+ messages=[Message("fixed partition")],
+ )
+
+ assert len(response.confirmations) == 1
+ assert response.confirmations[0].partition_id == partition_id
+
+ @pytest.mark.asyncio
+ async def test_send_messages_with_balanced_strategy(
Review Comment:
Let's change it like this: replace this test's single send with four sends using `Partitioning.balanced()`, collect the confirmation partition IDs, and assert `[0, 1, 2, 0]`.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "jiengup (via GitHub)" <gi...@apache.org>.
jiengup commented on PR #3927:
URL: https://github.com/apache/iggy/pull/3927#issuecomment-5479511009
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3892814229
##########
foreign/python/apache_iggy.pyi:
##########
@@ -2103,8 +2152,8 @@ class Topic:
r"""
Options admission resolved for the keys the client did not send.
- Same shape as `options`. These would have resolved differently under
- another server configuration.
+ Same shape as [`Self::options`]. These would have resolved differently
Review Comment:
Self::options is a rust term which python users will not understand.
##########
foreign/python/apache_iggy.pyi:
##########
@@ -2168,8 +2217,8 @@ class TopicDetails:
r"""
Options admission resolved for the keys the client did not send.
- Same shape as `options`. These would have resolved differently under
- another server configuration.
+ Same shape as [`Self::options`]. These would have resolved differently
+ under another server configuration.
Review Comment:
same as above -- `Self::options` is a rust term
##########
examples/python/README.md:
##########
@@ -11,7 +11,7 @@ To run any example, first start the server with
docker run --rm -p 8080:8080 -p 3000:3000 -p 8090:8090 apache/iggy:latest
# Or build from source (recommended for development)
-cd ../../ && cargo run --bin iggy-server
+cd ../../ && cargo run --bin iggy-server -- --with-default-root-credentials
Review Comment:
Let's make this `cargo run --bin iggy-server -- --with-default-root-credentials --fresh`
##########
foreign/python/src/partitioning.rs:
##########
@@ -0,0 +1,100 @@
+// 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 iggy::prelude::Partitioning as RustPartitioning;
+use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
+use pyo3_stub_gen::{
+ derive::{gen_stub_pyclass, gen_stub_pymethods},
+ impl_stub_type,
+};
+
+/// Defines how a batch of messages is assigned to a topic partition.
+#[derive(Clone)]
+#[pyclass(from_py_object)]
+#[gen_stub_pyclass]
+pub struct Partitioning {
+ pub(crate) inner: RustPartitioning,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Partitioning {
+ /// Routes the batch to one partition selected by round-robin.
+ #[staticmethod]
+ pub fn balanced() -> Self {
+ Self {
+ inner: RustPartitioning::balanced(),
+ }
+ }
+
+ /// Routes the batch to the specified partition.
+ ///
+ /// `partition_id` must be between 0 and `2**32 - 1`. The topic must contain
+ /// that partition when the batch is sent.
+ #[staticmethod]
+ pub fn partition_id(partition_id: u32) -> Self {
+ Self {
+ inner: RustPartitioning::partition_id(partition_id),
+ }
+ }
Review Comment:
`partition_id` accepts a Rust `u32`, so a Python integer outside 0 through `2**32 - 1` raises `OverflowError` during PyO3 conversion, while a non-integer raises `TypeError`. The public stub currently only describes the range, unlike `messages_key` which documents its failures. Add a `Raises` section so the constructor's contract matches `send_messages()`.
##########
examples/python/basic/producer.py:
##########
@@ -106,7 +106,11 @@ async def produce_messages(client: IggyClient):
await client.send_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
- partitioning=PARTITION_ID,
+ # A fixed strategy sends the whole batch to this partition.
+ # For topics with multiple partitions, Partitioning.balanced()
+ # distributes batches round-robin, while
+ # Partitioning.messages_key(key) keeps equal keys on the same partition.
+ partitioning=Partitioning.partition_id(PARTITION_ID),
Review Comment:
The linked issue explicitly requires a Python message-layer example covering all three strategies. Either edit the issue, or utilize all three strategies in the example.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3836019296
##########
foreign/python/apache_iggy.pyi:
##########
@@ -1601,6 +1605,30 @@ class Partition:
The number of messages in the partition.
"""
+@typing.final
+class Partitioning:
Review Comment:
`Partitioning` is missing from `__all__`, so from apache_iggy import * fails pyrefly with unknown-name.
##########
foreign/python/src/partitioning.rs:
##########
@@ -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.
+
+use iggy::prelude::Partitioning as RustPartitioning;
+use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
+use pyo3_stub_gen::{
+ derive::{gen_stub_pyclass, gen_stub_pymethods},
+ impl_stub_type,
+};
+
+/// Defines how a batch of messages is assigned to a topic partition.
+#[derive(Clone)]
+#[pyclass(from_py_object)]
+#[gen_stub_pyclass]
+pub struct Partitioning {
+ pub(crate) inner: RustPartitioning,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl Partitioning {
+ /// Routes the batch to partitions using server-side round-robin selection.
+ #[staticmethod]
+ pub fn balanced() -> Self {
+ Self {
+ inner: RustPartitioning::balanced(),
+ }
+ }
+
+ /// Routes the batch to the specified partition.
+ #[staticmethod]
+ pub fn partition_id(partition_id: u32) -> Self {
+ Self {
+ inner: RustPartitioning::partition_id(partition_id),
+ }
+ }
+
+ /// Routes the batch using a binary key hashed by the server.
+ ///
+ /// String keys are encoded as UTF-8. The encoded key must contain between
+ /// 1 and 255 bytes.
+ #[staticmethod]
+ pub fn messages_key(py: Python<'_>, key: PyMessagesKey) -> PyResult<Self> {
+ let key = match key {
+ PyMessagesKey::String(key) => key.into_bytes(),
+ PyMessagesKey::Bytes(key) => key.extract::<Vec<u8>>(py)?,
+ };
+ let inner = RustPartitioning::messages_key(&key)
+ .map_err(|error| PyValueError::new_err(error.to_string()))?;
+ Ok(Self { inner })
+ }
+}
+
+#[derive(FromPyObject)]
+pub enum PyMessagesKey {
+ #[pyo3(transparent, annotation = "str")]
+ String(String),
+ #[pyo3(transparent, annotation = "bytes")]
+ Bytes(Py<PyBytes>),
+}
+impl_stub_type!(PyMessagesKey = String | PyBytes);
+
+#[derive(FromPyObject)]
+pub(crate) enum PyPartitioning {
+ #[pyo3(transparent)]
+ Strategy(Partitioning),
Review Comment:
Strategy has no annotation, so errors read 'Strategy | int' — a Rust name that does not exist in Python. Please add annotation = "Partitioning"
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -109,6 +150,87 @@ async def test_send_messages_reports_committed_confirmation(
assert confirmation.partition_id == partition_id
assert confirmation.base_offset == polled_messages[0].offset()
+ @pytest.mark.asyncio
+ async def test_send_messages_with_partition_id_strategy(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 2
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=3
+ )
+
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=Partitioning.partition_id(partition_id),
+ messages=[Message("fixed partition")],
+ )
+
+ assert len(response.confirmations) == 1
+ assert response.confirmations[0].partition_id == partition_id
+
+ @pytest.mark.asyncio
+ async def test_send_messages_with_balanced_strategy(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partitions_count = 3
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=partitions_count,
+ )
+
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=Partitioning.balanced(),
+ messages=[Message("balanced")],
+ )
+
+ assert len(response.confirmations) == 1
+ assert response.confirmations[0].partition_id < partitions_count
Review Comment:
Sending once cannot tell round-robin apart from always returning partition 0. Could you send three batches and assert the three confirmed partition ids are distinct?
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(python): support message partitioning strategies [iggy]
Posted by "jiengup (via GitHub)" <gi...@apache.org>.
jiengup commented on code in PR #3927:
URL: https://github.com/apache/iggy/pull/3927#discussion_r3894994556
##########
examples/python/basic/producer.py:
##########
@@ -106,7 +106,11 @@ async def produce_messages(client: IggyClient):
await client.send_messages(
stream=STREAM_NAME,
topic=TOPIC_NAME,
- partitioning=PARTITION_ID,
+ # A fixed strategy sends the whole batch to this partition.
+ # For topics with multiple partitions, Partitioning.balanced()
+ # distributes batches round-robin, while
+ # Partitioning.messages_key(key) keeps equal keys on the same partition.
+ partitioning=Partitioning.partition_id(PARTITION_ID),
Review Comment:
And keep the comment here as a hint.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org