ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855709783
##########
foreign/python/src/consumer.rs:
##########
@@ -56,39 +62,33 @@ impl IggyConsumer {
/// Get the last consumed offset or `None` if no offset has been consumed
yet.
#[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
fn get_last_consumed_offset(&self, partition_id: u32) -> Option<u64> {
- self.inner
- .blocking_lock()
- .get_last_consumed_offset(partition_id)
+ self.state.get_last_consumed_offset(partition_id)
}
/// Get the last stored offset or `None` if no offset has been stored yet.
#[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
fn get_last_stored_offset(&self, partition_id: u32) -> Option<u64> {
- self.inner
- .blocking_lock()
- .get_last_stored_offset(partition_id)
+ self.state.get_last_stored_offset(partition_id)
}
/// Gets the name of the consumer group.
fn name(&self) -> String {
- self.inner.blocking_lock().name().to_string()
+ self.name.clone()
}
/// Gets the current partition id or `0` if no messages have been polled
yet.
fn partition_id(&self) -> u32 {
- self.inner.blocking_lock().partition_id()
+ self.state.partition_id()
}
/// Gets the name of the stream this consumer group is configured for.
Review Comment:
Reworded to identifier in 984494952
##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -856,6 +857,74 @@ async def test_consumer_group_metadata(self, iggy_client:
IggyClient, unique_nam
assert consumer.get_last_consumed_offset(partition_id) is None
assert consumer.get_last_stored_offset(partition_id) is None
+ @pytest.mark.asyncio
+ async def test_consumer_group_metadata_while_consuming(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test that metadata can be read while a consumption run is in
progress."""
+ consumer_name = unique_name()
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ message = f"Metadata test - {unique_name()}"
+ received_messages = []
+ consuming = asyncio.Event()
+ shutdown_event = asyncio.Event()
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name,
+ name=topic_name,
+ partitions_count=1,
+ )
+
+ consumer = await iggy_client.consumer_group(
+ consumer_name,
+ stream_name,
+ topic_name,
+ partition_id,
+ PollingStrategy.First(),
+ 10,
+ auto_commit=AutoCommit.Disabled(),
+ poll_interval=timedelta(milliseconds=25),
+ )
+
+ async def take(received: ReceiveMessage) -> None:
+ received_messages.append(received)
+ consuming.set()
+
+ await iggy_client.send_messages(
+ stream_name,
+ topic_name,
+ partition_id,
+ [Message(message)],
+ )
+
+ consume = consumer.consume_messages(take, shutdown_event)
+ try:
+ await asyncio.wait_for(consuming.wait(), timeout=10)
+
+ # A getter that blocks holds the GIL, so neither pytest-timeout
nor asyncio
+ # can fire. The faulthandler watchdog needs no GIL and aborts
instead.
+ # A regression hangs forever, so the timeout only has to outlast
normal GIL
+ # contention -- it is generous because tripping it kills the whole
run.
+ faulthandler.dump_traceback_later(5, exit=True)
Review Comment:
Moved to pyproject in c5aa65031: faulthandler_timeout +
faulthandler_exit_on_timeout.
--
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]
