[NOGIL] Restrict sharing of Consumer instances across threads - #2322
Open
Ojasva Jain (ojasvajain) wants to merge 6 commits into
Open
[NOGIL] Restrict sharing of Consumer instances across threads#2322Ojasva Jain (ojasvajain) wants to merge 6 commits into
Ojasva Jain (ojasvajain) wants to merge 6 commits into
Conversation
Ojasva Jain (ojasvajain)
requested review from
a team and
Matthew Seal (MSeal)
as code owners
August 6, 2026 13:33
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
airlock-confluentinc
Bot
force-pushed
the
dev_producer_no_gil
branch
from
August 14, 2026 07:33
015da26 to
4c750a2
Compare
…s protection Introduce a unified gate_owner/gate_depth mechanism in Consumer.c that rejects concurrent cross-caller access to a single Consumer instance while still allowing legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it). The gate is shared between the sync Consumer (thread ID as identity) and AIOConsumer (a Python-generated logical-caller identity, since the owning caller may move across ThreadPoolExecutor worker threads). Gate enforcement (CFL_CONSUMER_GATE_ENABLED in confluent_kafka.h) only applies to Python versions we've never shipped a wheel for -- 3.15+, or 3.14 built free-threaded -- so existing users on <=3.14 GIL-based Python see zero behavior change on upgrade. Also fixes Consumer__pause_internal/Consumer__resume_internal, which never checked self->rk before calling into librdkafka, causing pause()/resume() after close() to segfault instead of raising RuntimeError.
…endent txn race tests
…drop per-method wrappers
airlock-confluentinc
Bot
force-pushed
the
dev_consumer_no_gil
branch
from
August 14, 2026 07:39
3cd1566 to
da12753
Compare
Comment on lines
147
to
+196
| @@ -186,6 +190,10 @@ def worker(producer, stop_event): | |||
| def test_close_races_abort_transaction(): | |||
| """close() concurrent with abort_transaction() on another thread.""" | |||
|
|
|||
| # TODO NOGIL: move to tests/integration -- abort_transaction() needs a | |||
| # real transaction coordinator to reach the race being tested; against | |||
| # the unreachable localhost:9092 used here it fails with a genuine | |||
| # _STATE KafkaException instead of the expected RuntimeError. | |||
Member
Author
There was a problem hiding this comment.
Will revisit this in a separate PR that I plan to raise for Transactional Producer related audit.
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Introduces a reentrancy gate for the sync Consumer and AIOConsumer that rejects illegal concurrent access to a single Consumer instance from a different caller, while still allowing legitimate re-entrant calls — e.g. a rebalance or commit callback calling back into the Consumer/AIOConsumer that triggered it (on_assign calling assign(), on_commit calling commit(), etc.).
A unified gate_owner/gate_depth mechanism (Handle_gate_enter()/Handle_gate_exit() in Consumer.c) is shared between both sync and async consumer types, and every gated method resolves its own caller identity internally.
For sync consumer, calling thread ID serves as the identity while for the async consumer an ID is generated before dispatching each call and it gets carried over through the task chain using context variables. For async consumers, this is required as callbacks may get scheduled on a different worker thread, so a thread ID can not be used as an ID.
Backward compatibility: gate enforcement is controlled by CFL_CONSUMER_GATE_ENABLED (confluent_kafka.h) and only applies to Python versions this project has never shipped a wheel for — 3.15+ (any build), or 3.14 built free-threaded. On every version already shipped (≤3.14, GIL-based), the gate is compiled out entirely, so existing users see zero behavior change on upgrade.
TODO -> Add unit test cases for AIOConsumer