hubcio commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3851563522
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -486,8 +526,8 @@ impl IggyConsumer {
let consumer = self.consumer.clone();
let stream_id = self.stream_id.clone();
let topic_id = self.topic_id.clone();
- let last_consumed_offsets = self.last_consumed_offsets.clone();
- let last_stored_offsets = self.last_stored_offsets.clone();
+ let last_consumed_offsets = self.state.last_consumed_offsets.clone();
Review Comment:
the loop below holds a dashmap shard read guard across
`store_consumer_offset(...).await`, and the poll future can `insert` a new
partition into the same map (line 776). that insert blocks synchronously inside
an async task for a whole round trip - stalled worker normally, real deadlock
with a single worker (`#[tokio::test]` defaults to current_thread). collect
`(partition_id, offset)` into a vec first, then await.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -91,6 +91,47 @@ pub enum AutoCommitAfter {
ConsumingEveryNthMessage(u32),
}
+/// A cheap, cloneable view of the state shared with an [`IggyConsumer`].
+///
+/// Consuming borrows the consumer as `&mut` for the whole run, so reading its
getters
+/// concurrently means sharing it behind a lock and then waiting on that lock.
This view
+/// carries the same shared state and needs neither.
+#[derive(Clone)]
Review Comment:
public through the prelude with no `Debug` - `#[derive(Clone, Debug)]`
compiles. minor, `IggyConsumer` below has none either.
##########
core/sdk/src/clients/consumer.rs:
##########
@@ -91,6 +91,47 @@ pub enum AutoCommitAfter {
ConsumingEveryNthMessage(u32),
}
+/// A cheap, cloneable view of the state shared with an [`IggyConsumer`].
+///
+/// Consuming borrows the consumer as `&mut` for the whole run, so reading its
getters
+/// concurrently means sharing it behind a lock and then waiting on that lock.
This view
+/// carries the same shared state and needs neither.
+#[derive(Clone)]
+pub struct IggyConsumerState {
+ current_partition_id: Arc<AtomicU32>,
+ last_consumed_offsets: Arc<DashMap<u32, AtomicU64>>,
+ last_stored_offsets: Arc<DashMap<u32, AtomicU64>>,
+}
+
+impl IggyConsumerState {
+ fn new() -> Self {
+ Self {
+ current_partition_id: Arc::new(AtomicU32::new(0)),
+ last_consumed_offsets: Arc::new(DashMap::new()),
+ last_stored_offsets: Arc::new(DashMap::new()),
+ }
+ }
+
+ /// Returns the current partition ID of the consumer.
+ pub fn partition_id(&self) -> u32 {
Review Comment:
worth saying these aren't a combined snapshot - two independent loads, so
the partition id can be stale by the time you read an offset for it.
##########
foreign/python/src/consumer.rs:
##########
@@ -44,10 +44,16 @@ use crate::receive_message::ReceiveMessage;
/// A Python class representing the Iggy consumer.
/// It provides asynchronous functionality through the contained runtime.
+// `inner` stays locked for the whole duration of a consumption run, so the
synchronous
Review Comment:
`store_offset` and `delete_offset` still lock `inner`, so awaiting either
from inside a `consume_messages` callback deadlocks - the run holds the guard
while awaiting the callback. with `AutoCommit.Disabled()` that leaves no manual
commit path from a callback at all. both take `&self` on the rust side, so the
mutex isn't needed - same trick as the getters here.
##########
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.
Review Comment:
same - after any poll this returns 0 even with `AutoCommit.Disabled()` and
nothing stored server-side, which the new test at line 921 pins. don't change
the sentinel though, `store_consumer_offset` reads 0 as "nothing stored".
##########
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.
Review Comment:
`None` means the partition isn't tracked yet, not that nothing was consumed
- the poll future seeds 0 on first sight of a partition.
##########
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 {
Review Comment:
return `&str` and skip the clone - the generated stub comes out identical.
##########
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:
also line 89. these return `str | int`, not a name - a numeric-looking name
becomes a numeric id, so `consumer_group("g", "42", "t")` gives back the int
42. say identifier.
##########
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:
`exit=True` takes the whole session down, and there's no alternative - a
gil-pinned deadlock can't be recovered in process. pytest has it built in
though: `faulthandler_timeout` + `faulthandler_exit_on_timeout` in pyproject
gives the same thing suite-wide and drops these four lines.
--
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]
