You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "ethanlin01x (via GitHub)" <gi...@apache.org> on 2026/08/15 09:09:42 UTC
[PR] fix(python): read consumer metadata without the consumer lock [iggy]
ethanlin01x opened a new pull request, #3888:
URL: https://github.com/apache/iggy/pull/3888
## Which issue does this PR address?
Relates to #3776 (Found while reviewing)
## Rationale
The synchronous getters on the Python `IggyConsumer` took the consumer mutex with `blocking_lock()` while holding the GIL, and `consume_messages` holds that mutex for the whole consumption run. Reading an attribute during consumption hung the interpreter; reading one from a callback panicked inside the Tokio runtime. Neither is recoverable from Python.
```python
consume = consumer.consume_messages(handle, shutdown_event)
print(consumer.name()) # never returns
```
## What changed?
None of those getters need exclusive access. `name`, `stream` and `topic` are fixed at construction, so the Python wrapper now snapshots them. The partition id and offsets live behind `Arc`s in the Rust SDK, which had no way to hand them out.
The Rust SDK gains `IggyConsumerState`, a cheap cloneable view over that shared state, reachable via `IggyConsumer::state()`. `IggyConsumer` owns one and delegates its own `partition_id()` / `get_last_consumed_offset()` / `get_last_stored_offset()` to it, so there is a single implementation rather than two. The Python wrapper holds a clone and reads metadata without ever taking the lock.
Nothing in the Rust public API changed shape. On the Python side `stream()` and `topic()` no longer return a `PyResult`, since the conversion now happens once at construction; the generated stub is unchanged.
## Local Execution
- Passed
- Pre-commit hooks ran
## AI Usage
1. Which tools? Claude
2. Scope of usage? help implement and write PR description
3. How did you verify the generated code works correctly? Ran the Python test suite against a real server
4. Can you explain every line of the code if asked? Yes, all the changes are checked by the human.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5408709691
@slbotbm like we discussed on discord, please propose something via issues or discussions.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855694794
##########
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:
Doc updated in 984494952 to say these are independent loads, not a snapshot.
##########
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:
Both moved onto IggyConsumerState in b0d3c20c0, no lock taken from callbacks anymore.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5414539895
Also fixed a clippy error while at it.
For the apache_iggy.pyi CI gate, I'll dig into it and open a separate issue for discussion.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855703528
##########
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:
docs clarified in 984494952
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5301855857
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5301541249
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3888?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 82.44%. Comparing base ([`cbeb410`](https://app.codecov.io/gh/apache/iggy/commit/cbeb410f48dd63fffac0b0c33998c14768ad6c89?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8d0d07d`](https://app.codecov.io/gh/apache/iggy/commit/8d0d07d3c5590e95e6f5f250a0348e57643f562d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 5 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3888 +/- ##
============================================
- Coverage 82.89% 82.44% -0.45%
Complexity 1299 1299
============================================
Files 1200 1200
Lines 162348 162498 +150
Branches 131823 131784 -39
============================================
- Hits 134571 133971 -600
- Misses 24238 24985 +747
- Partials 3539 3542 +3
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.41% <ø> (+<0.01%)` | :arrow_up: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `66.06% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.97% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.92% <ø> (-0.07%)` | :arrow_down: |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.26% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.16% <ø> (-7.06%)` | :arrow_down: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3888/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.53% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3888?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/sdk/src/clients/consumer.rs](https://app.codecov.io/gh/apache/iggy/pull/3888?src=pr&el=tree&filepath=core%2Fsdk%2Fsrc%2Fclients%2Fconsumer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZGsvc3JjL2NsaWVudHMvY29uc3VtZXIucnM=) | `67.50% <ø> (+0.45%)` | :arrow_up: |
| [foreign/python/src/client.rs](https://app.codecov.io/gh/apache/iggy/pull/3888?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fclient.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2NsaWVudC5ycw==) | `99.84% <ø> (ø)` | |
| [foreign/python/src/consumer.rs](https://app.codecov.io/gh/apache/iggy/pull/3888?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fconsumer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2NvbnN1bWVyLnJz) | `81.17% <ø> (-1.05%)` | :arrow_down: |
| [foreign/python/src/identifier.rs](https://app.codecov.io/gh/apache/iggy/pull/3888?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fidentifier.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2lkZW50aWZpZXIucnM=) | `77.77% <ø> (ø)` | |
... and [61 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3888/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855693076
##########
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:
Added a manual Debug impl in 984494952 instead of derive, so the client lock and maps don't end up in the output.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #3888:
URL: https://github.com/apache/iggy/pull/3888
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5419226301
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "slbotbm (via GitHub)" <gi...@apache.org>.
slbotbm commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5408477672
@hubcio do we want to check the pyi file in ci? i ask because this check will require a complete recompilation of the sdk, and our ci runs are very heavy as it is.
maybe we could instead redesign the ci for ffi sdks instead?
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855702194
##########
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:
Docs and stubs clarified in 984494952
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on PR #3888:
URL: https://github.com/apache/iggy/pull/3888#issuecomment-5301854718
/request-review @hubcio
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855707073
##########
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:
Done in 984494952
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(python): read consumer metadata without the consumer lock [iggy]
Posted by "ethanlin01x (via GitHub)" <gi...@apache.org>.
ethanlin01x commented on code in PR #3888:
URL: https://github.com/apache/iggy/pull/3888#discussion_r3855690967
##########
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:
Fixed in 1ae5a4cc1 — collected into a Vec before awaiting.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org