slbotbm commented on code in PR #3877:
URL: https://github.com/apache/iggy/pull/3877#discussion_r3791217313
##########
foreign/python/tests/test_message_operations.py:
##########
Review Comment:
There are some more tests that I would like you to add:
- Create a topic with several partitions, put a distinguishable message in
each partition, join one client to the group, and repeatedly call
`poll_messages()` without a `partition_id`. The client should poll all of its
assigned partitions in round-robin order (assert this behaviour).
- Call `poll_messages()` with `Consumer.Single(...)` and no `partition_id`
after sending messages to partition `0`. The call should read partition `0`.
- Create a group member with a known assignment and call `poll_messages()`
with both `Consumer.Group(...)` and an explicitly owned `partition_id`. The
poll should return messages from that partition.
- Add small tests for the Python-to-Rust conversion boundary (some of these
could also be folded into existing tests):
- Omitting the required `consumer` argument should raise `TypeError`.
- Passing a value that is not a `Consumer` should raise `TypeError`.
- An invalid string identifier should raise `ValueError`.
- A negative integer or an integer larger than `u32` should be rejected.
- Valid numeric IDs should be accepted for both `Consumer.Single(...)`
and `Consumer.Group(...)`.
- If a client polls with `Consumer.Group(group_name)` without first joining
that group, it has no assigned partitions and is not allowed to consume for the
group. (should error)
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -1217,3 +1246,118 @@ async def
test_poll_messages_with_auto_commit_false_does_not_advance_next(
assert [
message.payload().decode("utf-8") for message in next_messages
] == existing_messages + new_messages
+
+ @pytest.mark.asyncio
+ async def
test_poll_messages_with_distinct_consumers_keeps_offsets_independent(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test each consumer owns its offset, so both see the whole
partition."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ test_messages = [f"Consumer isolation {i} - {unique_name()}" for i in
range(3)]
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(message) for message in test_messages],
+ )
+
+ for consumer_name in ("isolation-consumer-a", "isolation-consumer-b"):
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Single(consumer_name),
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Next(),
+ count=10,
+ auto_commit=True,
+ )
+ assert [
+ message.payload().decode("utf-8") for message in
polled_messages
+ ] == test_messages
+
+ @pytest.mark.asyncio
+ async def test_poll_messages_with_shared_consumer_splits_the_partition(
Review Comment:
`test_poll_messages_with_shared_consumer_splits_the_partition` ->
`test_poll_messages_with_shared_consumer_shares_the_partition`
--
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]
