hubcio commented on code in PR #4017: URL: https://github.com/apache/iggy/pull/4017#discussion_r3922790935
########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_create_and_delete_partitions(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=stream_name, name=topic_name, partitions_count=2 + ) + + await iggy_client.create_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 4 + assert [partition.id for partition in topic.partitions] == [0, 1, 2, 3] + + await iggy_client.delete_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 2 + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_accepts_numeric_ids( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + await iggy_client.create_topic( + stream=stream.id, name=topic_name, partitions_count=2 + ) + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + + await iggy_client.create_partitions(stream.id, topic.id, 1) + await iggy_client.delete_partitions(stream.id, topic.id, 1) + + topic = await iggy_client.get_topic(stream.id, topic.id) + assert topic is not None + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_rejects_zero_count( + 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=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.create_partitions(stream_name, topic_name, 0) + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, topic_name, 0) + + [email protected] +async def test_delete_partitions_rejects_count_larger_than_topic( + 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=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, topic_name, 3) + + [email protected] +async def test_partition_management_rejects_missing_stream_or_topic( Review Comment: nit: the documented `ValueError` for a bad identifier is never exercised here, and there's no delete-all-partitions case. skip an over-cap test though - zero and over-cap return the same code, so it would duplicate the zero-count assertion. ########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_create_and_delete_partitions(iggy_client: IggyClient, unique_name): Review Comment: nit: no test in this file ever sends a message, so delete is never exercised with data present and the new partitions are never shown to be usable. `test_topic.py:1258` is the shape to copy. don't assert topic `size_bytes` after the delete though - it isn't reclaimed. ########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_create_and_delete_partitions(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=stream_name, name=topic_name, partitions_count=2 + ) + + await iggy_client.create_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 4 + assert [partition.id for partition in topic.partitions] == [0, 1, 2, 3] + + await iggy_client.delete_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 2 + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_accepts_numeric_ids( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + await iggy_client.create_topic( + stream=stream.id, name=topic_name, partitions_count=2 + ) + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + + await iggy_client.create_partitions(stream.id, topic.id, 1) + await iggy_client.delete_partitions(stream.id, topic.id, 1) + + topic = await iggy_client.get_topic(stream.id, topic.id) + assert topic is not None + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_rejects_zero_count( + 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=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): Review Comment: warning: bare `pytest.raises(RuntimeError)` passes on any failure - every server error maps to `RuntimeError` here. add `match=`: "Too many partitions" for zero count, "Invalid partitions count" for over-count, "was not found." for missing stream/topic. also at lines 84, 100, 117, 119, 121, 123. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. + /// + /// Raises: + /// ValueError: If an identifier is invalid. + /// RuntimeError: If the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn create_partitions<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partitions_count: u32, + ) -> PyResult<Bound<'a, PyAny>> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .create_partitions(&stream_id, &topic_id, partitions_count) + .await + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; Review Comment: simplification: this makes 38 copies of the same `PyErr::new::<PyRuntimeError, _>(e.to_string())` map_err in this file. a `to_runtime_error(e: impl ToString)` helper, like `to_value_error` in `send_message.rs:93`, cuts each site to `.map_err(to_runtime_error)?`. also at line 843. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. Review Comment: warning: `partitions_count` has no documented range. server takes 1..=1000 (`MAX_PARTITIONS_PER_REQUEST`) and reports 0 as "Too many partitions", so spell it out here - the stub is the only doc a python caller gets. also at line 819. ########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_create_and_delete_partitions(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=stream_name, name=topic_name, partitions_count=2 + ) + + await iggy_client.create_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 4 + assert [partition.id for partition in topic.partitions] == [0, 1, 2, 3] + + await iggy_client.delete_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 2 + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_accepts_numeric_ids( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + await iggy_client.create_topic( + stream=stream.id, name=topic_name, partitions_count=2 + ) + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + + await iggy_client.create_partitions(stream.id, topic.id, 1) + await iggy_client.delete_partitions(stream.id, topic.id, 1) + + topic = await iggy_client.get_topic(stream.id, topic.id) + assert topic is not None + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_rejects_zero_count( + 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=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.create_partitions(stream_name, topic_name, 0) + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, topic_name, 0) + + [email protected] +async def test_delete_partitions_rejects_count_larger_than_topic( + 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=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): + await iggy_client.delete_partitions(stream_name, topic_name, 3) + + [email protected] +async def test_partition_management_rejects_missing_stream_or_topic( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + missing_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=2 + ) + + with pytest.raises(RuntimeError): Review Comment: simplification: these four raises blocks only differ in which identifier is missing. one `parametrize` over `(method, missing)` collapses them and keeps the test name. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. + /// + /// Raises: + /// ValueError: If an identifier is invalid. + /// RuntimeError: If the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn create_partitions<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partitions_count: u32, + ) -> PyResult<Bound<'a, PyAny>> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .create_partitions(&stream_id, &topic_id, partitions_count) + .await + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; + Ok(()) + }) + } + + /// Delete the last partitions from a topic, including all messages stored in them. Review Comment: nit: neither docstring mentions the consumer group side effect - delete rebalances members off the removed partitions and drops their stats, create leaves the new ones unassigned until the next rebalance. one line on each. also at line 780. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. + /// + /// Raises: + /// ValueError: If an identifier is invalid. + /// RuntimeError: If the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn create_partitions<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partitions_count: u32, + ) -> PyResult<Bound<'a, PyAny>> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .create_partitions(&stream_id, &topic_id, partitions_count) + .await + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; + Ok(()) + }) + } + + /// Delete the last partitions from a topic, including all messages stored in them. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to delete from the end of the topic. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are deleted. Review Comment: nit: the awaitable resolves on metadata commit, not after teardown - the unlink runs later in the reconciler and can back off. say "when the deletion is accepted; storage teardown completes asynchronously". ########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] Review Comment: nit: module-level test functions - 9 of the 10 other suites group into `Test*` classes. wrap these in `class TestPartitionManagement:`. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. + /// + /// Raises: Review Comment: nit: `Raises:` misses `OverflowError`, which pyo3 throws for a negative or oversized count and which is neither `ValueError` nor `RuntimeError`. `test_topic.py:367` already pins that behaviour. also at line 824. ########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_create_and_delete_partitions(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=stream_name, name=topic_name, partitions_count=2 + ) + + await iggy_client.create_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 4 + assert [partition.id for partition in topic.partitions] == [0, 1, 2, 3] + + await iggy_client.delete_partitions(stream_name, topic_name, 2) + topic = await iggy_client.get_topic(stream_name, topic_name) + assert topic is not None + assert topic.partitions_count == 2 + assert [partition.id for partition in topic.partitions] == [0, 1] + + [email protected] +async def test_partition_management_accepts_numeric_ids( + iggy_client: IggyClient, unique_name +): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + await iggy_client.create_topic( + stream=stream.id, name=topic_name, partitions_count=2 + ) + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + + await iggy_client.create_partitions(stream.id, topic.id, 1) + await iggy_client.delete_partitions(stream.id, topic.id, 1) + + topic = await iggy_client.get_topic(stream.id, topic.id) + assert topic is not None + assert [partition.id for partition in topic.partitions] == [0, 1] Review Comment: nit: this asserts `[0, 1]`, the same state as before the create and delete, so two no-ops that cancel out would pass. re-read between them: `created = await iggy_client.get_topic(stream.id, topic.id)`, then assert `created.partitions_count == 3`. ########## foreign/python/tests/test_partition.py: ########## @@ -0,0 +1,124 @@ +# 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 pytest + +from apache_iggy import IggyClient + + [email protected] +async def test_create_and_delete_partitions(iggy_client: IggyClient, unique_name): + stream_name = unique_name() Review Comment: simplification: the same 7-line create_stream + create_topic preamble is copy-pasted in four tests. pull it into a module-level helper like `test_consumer_group.py:37` - there is already a TODO there asking for exactly this. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. Review Comment: nit: nothing says which ids the new partitions get, and the caller needs them for `send_messages`. they're 0-based and append above the current max, reused after a delete - don't copy `partition_client.rs:26`, it claims 1-based and is wrong. ########## foreign/python/src/client.rs: ########## @@ -777,6 +777,74 @@ impl IggyClient { }) } + /// Create partitions for a topic. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to create. + /// + /// Returns: + /// An awaitable that resolves to `None` when the partitions are created. + /// + /// Raises: + /// ValueError: If an identifier is invalid. + /// RuntimeError: If the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn create_partitions<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partitions_count: u32, + ) -> PyResult<Bound<'a, PyAny>> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .create_partitions(&stream_id, &topic_id, partitions_count) + .await + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; + Ok(()) + }) + } + + /// Delete the last partitions from a topic, including all messages stored in them. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partitions_count: Number of partitions to delete from the end of the topic. Review Comment: nit: `partitions_count` drops the "as `int`" suffix the other args in this same docstring carry. adding it pushes the generated stub line past 88 chars, so rewrap rather than extend. also at line 785. -- 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]
