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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
