[NOGIL] Restrict sharing of Consumer instances across threads

[Ojasva Jain (ojasvajain)]

Introduces a reentrancy gate for the sync Consumer and AIOConsumer that serializes concurrent access to a single Consumer instance from a different caller — a colliding call waits for the current caller to finish rather than interleaving or racing directly against librdkafka's own consumer, which isn't thread-safe — while still allowing legitimate re-entrant calls to proceed immediately — 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 the sync consumer, calling thread ID serves as the identity, while for the async consumer an ID is generated before dispatching each call and 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 identity.

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Ojasva Jain (ojasvajain)]

Will revisit this in a separate PR that I plan to raise for Transactional Producer related audit.

[sonarqube-confluent]

Quality Gate failed

Failed conditions
8.4% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

[Kaushik Raina (k-raina)]

Thanks for PR!
Provided couple of correctness comments on gating logic.

[Kaushik Raina (k-raina)]

The gate identifies callers with a contextvar asyncio copies a task's context into every child task it spawns (gather, create_task), so any task launched from inside a callback inherits the same identity as the callback.

• When an application writes callbacks like:

  async def on_assign(consumer, partitions):
      await asyncio.gather(*(consumer.seek(tp) for tp in partitions))

  def on_commit(err, partitions):
      asyncio.create_task(consumer.commit())

• the spawned tasks inherit the callback's identity, so the owner == identity check in the gate is true for each of them and the gate admits them concurrently instead of serializing them.
• when poll() is running and its rebalance callback fires. The callback does gather(consumer.seek(a), consumer.seek(b)) two child tasks, both carrying the callback's copied identity are dispatched onto two executor worker threads.
• Task A enters Handle_gate_enter. Gate is free, so A takes it. A is now inside the C code, using self->rk
• Task B enters Handle_gate_enter. It sees gate_owner == identity (same copied badge), assumes it's a nested re-entrant call, does gate_depth++, and returns "admitted" immediately.
• Now A and B are both inside the gated code at once, both touching self->rk.

Since running on two threads at once, those updates can collide with each other. When both threads increment and decrement the same object's reference count without synchronization, one of the updates can be lost, and the recorded count drops below the number of references that actually exist. The object is then freed while it is still in use, and the next thread to reach for it dereferences memory that has already been released . The use after free that can crash the process.

Can we check if this can cause any issue?

[Ojasva Jain (ojasvajain)]

Yes I have documented it as a gap here(line 120). We can use python level lock to serialize the access. I will address this in a separate PR.

[Kaushik Raina (k-raina)]

Lets take an minimal example :

 async def on_assign(c, partitions):
      await c.assign(partitions)

  await consumer.subscribe([topic], on_assign=on_assign)

  await asyncio.gather(
      consumer.poll(10),
      consumer.commit(),
  )

AIOConsumer runs every call on a fixed thread pool (default 2 workers). A poll() takes worker 1 and the gate, then blocks waiting for its rebalance callback to return. A concurrent commit() (different thread) takes worker 2 and busy-spins at the gate instead of releasing its worker. The callback now makes a re-entrant call (assign/pause) that needs a worker but both are taken.

Application impact : For applications doing manual offset management (seek/commit from a rebalance callback) while any background task also calls the consumer will be stuck in Deadlock.

Could we check this?

[Ojasva Jain (ojasvajain)]

Yes this is possible if number of workers are less. But we can't have multiple tasks/threads touching the consumer handle at the same time and given AIO consumer schedules callbacks on different thread, we will need to accept this caveat. We will document this behaviour and recommend users to have enough workers in their thread pool.

Since this change will be part of a minor version upgrade, we decided to serialize the calls rather than returning ConcurrentModificationException to the users, as this will not be backward compatible. In the next major version upgrade, we can replace the waiting behaviour with throwing the exception which will avoid such deadlocks and change them to exceptions.

[Kaushik Raina (k-raina)]

For code snippet:

async def consume_loop():
    while running:
        msg = await consumer.poll(-1)     # block until the next message
        if msg:
            await process(msg)

async def shutdown_on_signal():
    await consumer.close()                # ← must wait behind poll(-1) at the gate

await asyncio.gather(consume_loop(), shutdown_on_signal())

1. consume_loop's poll(-1) calls Handle_gate_enter and takes the gate, then enters a chunk loop. With timeout = -1, poll forever until a message arrives. The gate is held the entire time.
2. shutdown's close() calls Handle_gate_enter, sees the gate owned by poll (different identity), and spins in the wait loop (usleep) waiting for poll to release it.

Application impact
For applications with Idle topic → no message → poll(-1) never returns → the gate is never released → close spins forever. The consumer is stuck.

Can we check this?

[Ojasva Jain (ojasvajain)]

Yes. This will expected. And this is the cost of serializing the calls, rather then returning the exception but that will not be backward compatible.

In the latest version of the python client, calling close() in parallel with poll(-1) is unsafe and causes hangs or segfault. So this is not a supported pattern anyway.

In Java client, close() will get a ConcurrentModificationException.