[NOGIL] Fix Producer.close() races with concurrent calls and with itself by ojasvajain · Pull Request #2313 · confluentinc/confluent-kafka-python

Ojasva Jain (ojasvajain) · 2026-07-21T13:52:19Z

Producer.close() previously raced with concurrent produce()/poll()/ flush()/produce_batch()/transaction calls and with itself when called from multiple threads, both leading to use-after-free/double-free on the underlying rd_kafka_t handle.

This PR adds:

An active_calls/closing guard (Handle_enter_rk_use/Handle_exit_rk_use) so every method that touches self->rk registers itself before use, and close() drains in-flight calls before tearing down;
A CAS on closing ensures only one concurrent close() call performs the actual teardown, with losing callers getting a False as return value and a warning.
Unit and integration test cases

What

Checklist

Contains customer facing changes? Including API/behavior changes
Did you add sufficient unit test and/or integration test coverage for this PR?
- If not, please explain why it is not required

References

JIRA:

Test & Review

Open questions / Follow-ups

confluent-cla-assistant · 2026-07-21T13:52:30Z

🎉 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.}

Kaushik Raina (k-raina)

Thanks for PR!
I have reviewed correctness and scope of PR, in this round.

Please check below methods which needs NOGIL gaurds

set_sasl_credentials
list_topics

these methods are defines in Metadata.c

Kaushik Raina (k-raina) · 2026-07-29T07:31:30Z

+#else
+                        usleep(100000);
+#endif
+                        CallState_end(self, &cs);


CallState_end returns 0 when PyErr_CheckSignals() fired or a callback crashed — and it leaves a Python exception set.
Should we check return value of this?

Yes, I think we should check the value of PyErr_CheckSignals. Not required for callback crash because the thread will only call sleep so callback crash can not happen. Will try to also add a test case.

Fixed this and added a test case.

Kaushik Raina (k-raina) · 2026-07-29T07:33:39Z

+
+
+def test_close_waits_for_in_flight_call():
+    """close() blocks until an in-flight poll() call finishes."""


Currently default for flush(-1) or poll(-1) runs forever, if close blocks on this. Is that usecase left intentionally?

Valid concern. In poll() and flush, we use chunked polling and check for signals in between. I have modified it to also check if close() has been called or not. If yes, poll() / flush() will exit early. In close(), we are anyway doing a indefinite flush before destroying rk.

Kaushik Raina (k-raina) · 2026-07-29T07:34:55Z

if user code calls producer.close() from inside a delivery-report callback, that callback is running inside poll()/flush(), i.e. inside that same thread's active_calls bump so close() waits for a call that is waiting on close(). Self-deadlock on one thread. How are we handling this usecase?

Thought about this. Even in current version, calling close() from inside a callback is not supported (the client will simply segfault because after close() completes, rk will be NULL). After my changes, calling a close from cb will cause a deadlock instead of a segfault. There are ways to detect such deadlocks and workaround them but the implementation is not trivial so I am not sure it should be in the scope of free threading changes. We can document that calling close() from a cb will cause a deadlock so it is not recommended.

Can we add TODO, to remember adding it to documentation ?

Added a TODO in Producer_close()

Kaushik Raina (k-raina) · 2026-07-29T07:50:07Z


-        if (!(c_offsets = py_to_c_parts(offsets)))
+        if (!(c_offsets = py_to_c_parts(offsets))) {
+                Handle_exit_rk_use(self);


Earch exist branch calls Handle_exit_rk_use. Wondering if we could use "goto exit:" semantics to future proof codebase from "new branch missing Handle_exit_rk_use calls" bugs?

Eg. go to pattern is used in other parts of codebase

confluent-kafka-python/src/confluent_kafka/src/Producer.c

Lines 891 to 905 in 85f31c7

cleanup:

Handle_exit_rk_use(self);

/* Cleanup resources */

if (rkt)

rd_kafka_topic_destroy(rkt);

if (rkmessages)

free(rkmessages);

if (msgstates)

free(msgstates);

if (PyErr_Occurred())

return NULL;

return cfl_PyInt_FromInt(good);

I think it makes sense to add it in transaction related methods where there can be multiple exit points. I will refactor them, taking care that we don't introduce a regression.

Kaushik Raina (k-raina) · 2026-07-29T07:51:55Z

+"""
+Shared test infrastructure for tests/concurrency/.
+
+Tests here deliberately race Producer/Consumer methods against each other,


Nice optimisation!

Kaushik Raina (k-raina) · 2026-07-29T08:05:15Z

+#if defined(_MSC_VER)
+typedef volatile LONG atomic_int_t;
+
+#define atomic_int_init(p, v) (*(p) = (v))


Seems like dead code?

No, these are windows implementations of the atomic APIs. The are defined in #if defined(_MSC_VER).

Kaushik Raina (k-raina) · 2026-07-29T08:17:12Z

+        Handle_exit_rk_use(self);
+
        /* Cleanup resources */
        if (rkt)


A concurrent close() frees rk, and rd_kafka_topic_destroy will use rkt->rkt_rk. Can this cause concurrency bug?

Nice catch. Yes, this call needs to be added after the rkt destroy.

Kaushik Raina (k-raina) · 2026-07-29T08:35:14Z

        CallState_begin(self, &cs);

        /* Flush any pending messages (wait indefinitely to ensure delivery) */
        err = rd_kafka_flush(self->rk, -1);


If callback touches the producer, it hits Handle_enter_rk_use(), sees closing == 1, and raises RuntimeError: Producer closed.

That Error propagates into librdkafka's callback dispatch, which calls rd_kafka_yield.

So the flush aborts early, close() raises, and messages still in the queue are dropped.

Could we verify, if flushing is able to flush all messages?

Added a test case to check if all msgs get flushed. As mentioned before, we don't want to support callbacks with close() calls.

Ojasva Jain (ojasvajain) · 2026-07-29T14:52:19Z

Please check below methods which needs NOGIL gaurds

set_sasl_credentials

list_topics

these methods are defines in Metadata.c

Yes, these are out of scope for this PR. I am addressing them in a separate PR as they are common to all clients.

Kaushik Raina (k-raina)

Left couple of comments on code correctness.

Kaushik Raina (k-raina) · 2026-08-03T17:21:56Z

+                producer.init_transactions(0.05)
+            except RuntimeError:
+                break
+            except Exception:  # noqa: BLE001 - librdkafka state/timeout errors are expected without a broker


Should we pass in this case? It has potential to miss real bugs?

Yes, have removed them.

Kaushik Raina (k-raina) · 2026-08-03T17:49:52Z

+        if (!atomic_int_cas(&self->closing, 0, 1)) {
+                PyErr_WarnFormat(PyExc_RuntimeWarning, 1,
+                                 "Producer is already closing");
+                Py_RETURN_FALSE;


IIUC,
Lets take example of code :

with Producer(config) as producer: producer.produce("topic", b"important")

If there are 2 threads:
T1: with clause producer calls close(), gets lock and passed to flush and destroy.
T2 : producer calls close(), checks cas and immediately returns false. As per current definition, false means "Context manager exit. Automatically flushes and destroys the producer."

So applications running T2 will assume producer is closed sucessfully with all messages flushed. However that is not true, as T1 is flushing which is an expensive call. This will fail applications which are running below usecases:

If the loser is the thread holding the program open, the process can exit mid-flush. Hence losing messages. This is common usecase for containerized service receiving SIGTERM

If loser is following up with another cleanup at application layer. eg. applications with lifecyle hooks.

This seems to be more frequently hit bug in multi threaded environment.

Hmm. Since we will now document that close() from a callback is not supported and will cause deadlocks, it should be safe to add the logic for waiting for the close call to complete.

Since we will now document that close() from a callback is not supported and will cause deadlocks

Above mentioned issue will occur with two different threads calling producer.close() separately. Its unrelated to callbacks.

Yes but supporting BOTH 1) calling close() from callbacks and 2) other threads waiting for close to complete, is not straightforward. Now that we are not supporting 1, we can support 2.

Are we supporting close from producer delivery callbacks?

No, we will not support it. This is not supported today as well, calling close() from within a callback causes a seg fault, so there can not be existing users calling close() from callbacks. After my changes, seg fault now changes to a deadlock. We will document this behaviour.

Kaushik Raina (k-raina) · 2026-08-03T18:14:52Z

+                usleep(100000);
+#endif
+                if (!CallState_end(self, &cs))
+                        return NULL;


Returning null leaves rk alive, as rd_kafka_destroy(self->rk); is never reached. Does this need to be fixed?

Yes. It leaves rk in an inconsistent state where it is not null and closing is set to 1, so any future call on the client will fail, despite rk being alive. Here's what I am thinking:

When we return NULL here, we also set closing back to 0, so any future call suceeds.

If other threads were waiting for the close to complete and the winner got interrupted and returned NULL, (ref), the other threads should also get an exception.

Done, also added a test case.

Kaushik Raina (k-raina) · 2026-08-03T18:18:17Z

+###############################################################################
+
+
+def test_close_completes_quickly_with_indefinite_poll_in_progress():


Does these tests, added in new commi, also needs to be isolated?

I have isolated only those test cases where a segmentation fault is possible.

Kaushik Raina (k-raina) · 2026-08-03T18:18:51Z

+        assert any(all_results), f"iteration {i}: expected at least one close() call to return True, got: {all_results}"
+
+
+###############################################################################


Is this seperator needed? It doesn't have any description of what it seperates

This was meant to mark the end of test cases where we are racing close() vs other methods. Will make it more clear.

sonarqube-confluent · 2026-08-04T15:41:51Z

Quality Gate failed

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

See analysis details on SonarQube

Kaushik Raina (k-raina)

Thanks for addressing comments. Provided few more comments, mainly on correctness.

Kaushik Raina (k-raina) · 2026-08-07T12:51:15Z

+        return result;
 }

 static PyObject *Producer_abort_transaction(Handle *self, PyObject *args) {


When a transaction hits an abortable error, librdkafka returns via err.txn_requires_abort(). Python then calls abort_transaction(). However closing latch makes every transaction API raise RuntimeError as soon as any thread starts closing including abort_transaction().

Impact on application
Once any thread starts closing a transactional producer, every transaction call — including the mandatory abort — is refused for good, and close() doesn't perform the abort itself. Although, The coordinator holds the transaction open until transaction.timeout.ms expires (default 60 s) then aborts it server-side.
So the records land, stay invisible, and are eventually discarded.

Maybe we should think about if there can be any issue here?

Yes, there could be an issue. I need to understand how it is handled in existing impl and Java client. I am adding a TODO to revisit this and Tx Producer in general. In this PR, I only meant to verify the basic working of Tx Producer by adding test cases.

Kaushik Raina (k-raina) · 2026-08-13T06:02:49Z

+    Note: calling close() from within a callback is NOT covered here and is
+    NOT supported."""
+
+    def test_delivery_callback_producing_another_message_gets_delivered(self, kafka_cluster):


One usecase which we need to think of:

When close is called, rd_kafka_flush(rk, -1) dispatches delivery report dr_msg_cb()

Callback calls produce(), which hits the gate at Producer.c:305, sees closing == 1, and raises RuntimeError: Producer has been closed.

In dr_msg_cb the callback returning an exception is treated as a fatal event https://github.com/confluentinc/confluent-kafka-python/blob/master/src/confluent_kafka/src/Producer.c#L134-L178

rd_kafka_yield call bail out of whatever blocking call is in progress on this thread so flush(-1), which is supposed to wait forever for delivery, returns after a few milliseconds with the queue still full.

CallState_end() then re-raises the stashed exception, so close() raises RuntimeError: Producer has been closed.

We can try to remove producer.flush(30) and check if it this case can happen? We might have to validate it by adding tests or logical workflow.

If this bug is possible, Does it means we cannot add any producer call from inside callback?

Nice catch! This bug will surface when a delivery callback contains a producer API call and the callback gets called by flush(-1) during close. I am fixing it by introducing a new field called closing_thread that will be used in Handle_enter_rk_use to detect if it is a re-entrant call by the same thread or not. If yes, we allow the call to happen. If it is a different thread, we raise the exception. Also adding a test case.

Is this fix complete?
For closing thread itself, reentrantly, from a delivery callback that close()'s flush is dispatching. If poll is called from same thread that close() calbacks are running. They will be abruptly retuned because poll and flush doesn;t have check for closing_thread.

The fix is complete and the above behaviour is intended. Whatever user intends to do by calling poll() or flush() in a delivery callback, would be taken care by flush(-1) being called in close().

Kaushik Raina (k-raina) · 2026-08-13T07:01:27Z

confluent-kafka-python/src/confluent_kafka/src/Producer.c

Lines 1542 to 1554 in ba1020c

static int Producer_init(PyObject *selfobj, PyObject *args, PyObject *kwargs) {

Handle *self = (Handle *)selfobj;

char errstr[256];

rd_kafka_conf_t *conf;

if (self->rk) {

PyErr_SetString(PyExc_RuntimeError,

"Producer already __init__:ialized");

return -1;

}

self->type = RD_KAFKA_PRODUCER;

Do we need to update this code to check closing also?

No. A producer can't be closed when a producer instance is being created.

Kaushik Raina (k-raina) · 2026-08-13T07:34:28Z

+ */
+int Handle_enter_rk_use(Handle *h) {
+        if (atomic_int_get(&h->closing) || !h->rk) {
+                PyErr_SetString(PyExc_RuntimeError, ERR_MSG_PRODUCER_CLOSED);


We should check if we need to add/update documentation for API regarding RuntimeError

The runtime errors have always been there. In this PR, we are only hardening the close check, not introducing it. I do plan to update the docs once all implementation and testing PRs are merged. As part of that, I will see if we should document this behaviour explicitly.

Ojasva Jain (ojasvajain) · 2026-08-13T10:48:18Z

+#if defined(_MSC_VER)
+typedef volatile LONG_PTR atomic_ulong_t;
+
+#define atomic_ulong_init(p, v) (*(p) = (v))
+#define atomic_ulong_get(p)                                                   \
+        ((unsigned long)InterlockedCompareExchangePointer(                   \
+            (PVOID volatile *)(p), 0, 0))
+#define atomic_ulong_set(p, v)                                                \
+        InterlockedExchangePointer((PVOID volatile *)(p), (PVOID)(v))
+
+#else /* gcc / clang */
+typedef unsigned long atomic_ulong_t;
+
+#define atomic_ulong_init(p, v) __atomic_store_n((p), (v), __ATOMIC_SEQ_CST)
+#define atomic_ulong_get(p) __atomic_load_n((p), __ATOMIC_SEQ_CST)
+#define atomic_ulong_set(p, v) __atomic_store_n((p), (v), __ATOMIC_SEQ_CST)
+#endif


FYI - These APIs are also needed in the Consumer PR.

Producer.close() previously raced with concurrent produce()/poll()/ flush()/produce_batch()/transaction calls and with itself when called from multiple threads, both leading to use-after-free/double-free on the underlying rd_kafka_t handle. Adds an active_calls/closing guard (Handle_enter_rk_use/Handle_exit_rk_use) so every method that touches self->rk registers itself before use, and close() drains in-flight calls before tearing down; a CAS on `closing` ensures only one concurrent close() call performs the actual teardown, with losing callers waiting for it to finish rather than racing it. Adds tests/parallel/test_producer_close_race.py covering each affected method racing close(), close() racing itself, and close()'s blocking behavior. Uses pytest-forked (POSIX only) so a regression segfault fails only that test. Integration tests against a real broker are still pending.

…for Producer close()/transaction races

…ancy tests Concurrent close() calls now return False immediately with a warning instead of waiting for the CAS winner, since waiting could deadlock a caller that already holds an active_calls slot (e.g. a callback invoked from its own poll()/flush()). poll()/flush() now also exit early once closing is set instead of blocking the drain-wait. Fixes an ordering bug in produce_batch() where the topic handle was destroyed after releasing the active_calls slot. Adds integration tests for reentrant callbacks and close()'s internal flush delivering all messages, and documents close()-from-callback as unsupported.

Kaushik Raina (k-raina)

More comments on correctness

Kaushik Raina (k-raina) · 2026-08-16T16:40:28Z

+
+        producer.close()
+
+    def test_delivery_callback_calling_poll_does_not_crash_or_hang(self, kafka_cluster):


This test is claiming to applications that calling a poll from delivery callback is safe.

Create a Producer and register a delivery callback that calls poll(0) .

produce() more than one message with that callback (a real batch, this test created one message).

A delivery report fires callback; the callback calls poll(0).

That nested poll(0) releases the GIL and pushes a nested CallState.

On exit it clears the handle's thread-local slot to NULL instead of restoring it.

When subsequent delivery callback fires and reads the now-NULL slot.

CallState_get dereferences NULL → hard SIGSEGV

Can we try out this to verify claim?

Yeah tried it and it fails. The segmentation fault is caused because of a TLS corruption and occurs on master branch too. Since it is a pre-existing bug, the fix should not be in scope of free threading changes. As for this test case, I have updated the docstring and renamed the test case to better express what's inteded to be tested.

I'm seeing the same issue when calling flush() from a delivery callback.

I couldn't find any documentation clarifying whether calling poll() or flush() from a delivery callback is supported. From an application perspective, supporting these calls may not make sense: they may already have been invoked by the application, or the client may already be draining the callback state.

Could we verify the documentation and clarify whether poll() and flush() should be supported from delivery callbacks?

I'm seeing the same issue when calling flush() from a delivery callback.

yes that's expected. poll() and flush() are similar.

Could we verify the documentation and clarify whether poll() and flush() should be supported from delivery callbacks?

I don't think it is documented anywhere but what we know is that Python client has never supported it. As part of our documentation, we can add a section about dos and donts about callbacks. While working on this PR, we have figured out many things that were not supported in callbacks so we can explicitly mention about them in our docs.

Kaushik Raina (k-raina) · 2026-08-17T07:43:28Z

+    Note: calling close() from within a callback is NOT covered here and is
+    NOT supported."""
+
+    def test_delivery_callback_producing_another_message_gets_delivered(self, kafka_cluster):


Is this fix complete?
For closing thread itself, reentrantly, from a delivery callback that close()'s flush is dispatching. If poll is called from same thread that close() calbacks are running. They will be abruptly retuned because poll and flush doesn;t have check for closing_thread.

mcr-kaes · 2026-08-17T11:44:56Z

unsubscribe From: sonarqube-confluent[bot] ***@***.***> Date: Monday, August 17, 2026 at 6:19 AM To: confluentinc/confluent-kafka-python ***@***.***> Cc: Subscribed ***@***.***> Subject: Re: [confluentinc/confluent-kafka-python] [NOGIL] Fix Producer.close() races with concurrent calls and with itself (PR #2313) Sent by an external sender

…

Kaushik Raina (k-raina)

Overall LGTM, couple of test hardening comments

Kaushik Raina (k-raina) · 2026-08-18T06:17:06Z

+    while not stop_event.is_set():
+        try:
+            producer.produce('mytopic', value=b'x')
+        except RuntimeError:


produce() raises BufferError and transaction APIs raise KafkaException. Should we include these exceptions in catch to prevent test run crash?

The test will not crash if produce() returns any unexpected exceptions because the worker call is wrapped in another try catch (ref). The catch there is a general purpose catch block which will append the errors and then we have an assertion on errors being empty.

Kaushik Raina (k-raina) · 2026-08-18T06:54:01Z

+        try:
+            producer.produce('mytopic', value=b'x')
+        except RuntimeError:
+            # Expected once close() has fully completed on this thread's


Workers catch the expected RuntimeError and break silently , so rejection is not asserted.

_race_close_against tests run worker threads that repeatedly call producer APIs while close() executes on the main thread, then assert that no worker hung, no unexpected exception escaped, and every close() returned True.

Do we need to update Exception path case?

Workers catch the expected RuntimeError and break silently , so rejection is not asserted.

Yeah, this can probably be hardened. Will push the fix in some time.

then assert that no worker hung, no unexpected exception escaped, and every close() returned True.

We are not checking for hangs, only unexpected exceptions and close() return value.

Pushed a commit that hardens the race checks.

sonarqube-confluent · 2026-08-18T10:24:12Z

Quality Gate failed

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

See analysis details on SonarQube

Kaushik Raina (k-raina)

LGTM! Thanks for PR. Please plan for the following issues mentioned in review:

poll(), flush(), and close() are unsupported within C callbacks.
Transaction handling for free-threaded applications.

Ojasva Jain (ojasvajain) · 2026-08-18T13:18:24Z

poll(), flush(), and close() are unsupported within C callbacks.

I will document this.

Transaction handling for free-threaded applications.

Yes. This will be part of a later PR.

…elf (#2313) * Fix Producer.close() races with concurrent calls and with itself Producer.close() previously raced with concurrent produce()/poll()/ flush()/produce_batch()/transaction calls and with itself when called from multiple threads, both leading to use-after-free/double-free on the underlying rd_kafka_t handle. Adds an active_calls/closing guard (Handle_enter_rk_use/Handle_exit_rk_use) so every method that touches self->rk registers itself before use, and close() drains in-flight calls before tearing down; a CAS on `closing` ensures only one concurrent close() call performs the actual teardown, with losing callers waiting for it to finish rather than racing it. Adds tests/parallel/test_producer_close_race.py covering each affected method racing close(), close() racing itself, and close()'s blocking behavior. Uses pytest-forked (POSIX only) so a regression segfault fails only that test. Integration tests against a real broker are still pending. * Replace pytest-forked with subprocess-based test isolation * Fix subprocess_isolated import and flaky close() timing assertion * Rename tests/parallel to tests/concurrency and add integration tests for Producer close()/transaction races * Clarified comment * Make Producer.close() non-blocking for concurrent callers, add reentrancy tests Concurrent close() calls now return False immediately with a warning instead of waiting for the CAS winner, since waiting could deadlock a caller that already holds an active_calls slot (e.g. a callback invoked from its own poll()/flush()). poll()/flush() now also exit early once closing is set instead of blocking the drain-wait. Fixes an ordering bug in produce_batch() where the topic handle was destroyed after releasing the active_calls slot. Adds integration tests for reentrant callbacks and close()'s internal flush delivering all messages, and documents close()-from-callback as unsupported. * Fix close() losers to wait for winner instead of returning False early * Fix CI flakiness in test_close_races_close_losers_wait_for_slow_winner * Move signal/slow-winner close() race tests to integration suite * Allow reentrant Producer calls from close()'s own delivery callback * Update docstring of test case * Harden race checks

* [NOGIL] Add CI verification jobs for free-threaded Python 3.14t (#2304) * Add CI verification jobs for free-threaded Python 3.14t - Add a Semaphore block running source package verification and integration tests on CPython 3.14t (classic and consumer group protocols). cimpl does not declare free-threading support yet, so importing it re-enables the GIL: these jobs validate the 3.14t toolchain and packaging until that declaration ships. - Skip the CI-only orjson install on free-threaded interpreters: no free-threaded orjson wheels exist and the source build would fail; the stdlib JSON fallback path stays covered. - Add a module-scoped autouse fixture (defined on free-threaded builds only) that warns when the GIL is re-enabled around a test module. Hard asserts are staged behind TODO FTS markers, to be enabled in the same PR that declares Py_MOD_GIL_NOT_USED. Interpreter detection follows the free-threading HOWTO: https://docs.python.org/3/howto/free-threading-python.html * Handle deps without free-threaded wheels in test setup On free-threaded (no-GIL) builds, the rules and json-fast extras' compiled deps (tink, google-re2, grpcio; orjson) ship no free-threaded wheels and fail to build from source, so: - Add requirements-tests-install-nogil.txt, a variant of requirements-tests-install.txt without those extras, and install it from source-package-verification.sh when the interpreter is free-threaded (detected via Py_GIL_DISABLED). - Exclude the schema_registry test modules that import tink/celpy/orjson at the top of the file from collection on free-threaded builds only; on regular builds a missing dep stays a loud collection error rather than a silent skip. Plain serdes coverage recovery is marked as a TODO NOGIL follow-up. * Style fixes * [NOGIL] Split schema_registry tests to narrow free-threaded test exclusions (#2309) * [NOGIL] Split schema_registry tests to narrow free-threaded test exclusions Each of test_avro_serdes.py, test_config.py, test_json_serdes.py, and test_proto_serdes.py is split into a plain file (no rules/encryption dependency) and a _rules file (CEL/encryption/JSONata-dependent tests). This lets the plain tests run on free-threaded (3.14t) builds, where tink/celpy/orjson have no free-threaded wheels, while only the _rules files stay excluded via conftest.py's collect_ignore. * Fix isort/black formatting in split schema_registry tests * [NOGIL] Fix post-rebase gap: exclude azure/tink-dependent schema_registry tests from free-threaded collection test_azure_aead.py, test_azure_client.py, test_azure_driver.py, and test_encrypt_executor.py import azure/tink unconditionally at module level but were missing from conftest.py's free-threaded collect_ignore list, causing collection errors on 3.14t CI. These tests were added to master after the original NOGIL schema_registry exclusion list (#2309) landed, so they weren't accounted for; rebasing onto master surfaced the gap. * [NOGIL] Fix post-rebase gap: update schema_registry free-threaded exclusions test_json.py's orjson import became pytest.importorskip upstream, so it no longer needs excluding. test_dlq_serdes.py is new post-rebase and hard- imports celpy, so it needs excluding. * [NOGIL] Exclude test_dlq.py from free-threaded schema_registry integration test collection * [NOGIL] Fix Producer.close() races with concurrent calls and with itself (#2313) * Fix Producer.close() races with concurrent calls and with itself Producer.close() previously raced with concurrent produce()/poll()/ flush()/produce_batch()/transaction calls and with itself when called from multiple threads, both leading to use-after-free/double-free on the underlying rd_kafka_t handle. Adds an active_calls/closing guard (Handle_enter_rk_use/Handle_exit_rk_use) so every method that touches self->rk registers itself before use, and close() drains in-flight calls before tearing down; a CAS on `closing` ensures only one concurrent close() call performs the actual teardown, with losing callers waiting for it to finish rather than racing it. Adds tests/parallel/test_producer_close_race.py covering each affected method racing close(), close() racing itself, and close()'s blocking behavior. Uses pytest-forked (POSIX only) so a regression segfault fails only that test. Integration tests against a real broker are still pending. * Replace pytest-forked with subprocess-based test isolation * Fix subprocess_isolated import and flaky close() timing assertion * Rename tests/parallel to tests/concurrency and add integration tests for Producer close()/transaction races * Clarified comment * Make Producer.close() non-blocking for concurrent callers, add reentrancy tests Concurrent close() calls now return False immediately with a warning instead of waiting for the CAS winner, since waiting could deadlock a caller that already holds an active_calls slot (e.g. a callback invoked from its own poll()/flush()). poll()/flush() now also exit early once closing is set instead of blocking the drain-wait. Fixes an ordering bug in produce_batch() where the topic handle was destroyed after releasing the active_calls slot. Adds integration tests for reentrant callbacks and close()'s internal flush delivering all messages, and documents close()-from-callback as unsupported. * Fix close() losers to wait for winner instead of returning False early * Fix CI flakiness in test_close_races_close_losers_wait_for_slow_winner * Move signal/slow-winner close() race tests to integration suite * Allow reentrant Producer calls from close()'s own delivery callback * Update docstring of test case * Harden race checks * [NOGIL] Restrict sharing of Consumer instances across threads (#2322) * Add a serializing Consumer reentrancy gate for free-threading support Introduces gate_owner/gate_depth in Consumer.c so concurrent, cross-caller access to a single Consumer/AIOConsumer instance waits for the current caller to finish rather than being undefined behavior, since librdkafka's consumer is not thread-safe; legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it) are still admitted immediately. AIOConsumer identity is tracked via a ContextVar since the owning logical caller can move across ThreadPoolExecutor worker threads. Includes unit and integration test coverage for both the sync Consumer and AIOConsumer. * Style fixes * Fix flaky tests * Add TODO for handling concurrent calls inside callback * Addressed comments * Trigger CLA check * Address comments * Return consistent error when consumer is closed * Fix one test case * Fix styling * [NOGIL] Exclude Avro tests from running on free threaded python builds (#2335) * Exclude Avro tests from running on free threaded python builds * Exclude only fastavro * temoporarily install fastavro for docs * [NOGIL] Declare thread free support * [NOGIL] Improve Test Coverage for Producer + Some Edge Case Fixes (#2329) * Add a serializing Consumer reentrancy gate for free-threading support Introduces gate_owner/gate_depth in Consumer.c so concurrent, cross-caller access to a single Consumer/AIOConsumer instance waits for the current caller to finish rather than being undefined behavior, since librdkafka's consumer is not thread-safe; legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it) are still admitted immediately. AIOConsumer identity is tracked via a ContextVar since the owning logical caller can move across ThreadPoolExecutor worker threads. Includes unit and integration test coverage for both the sync Consumer and AIOConsumer. * Trigger CLA check * Add Integration test coverage for AIO Producer * Add Tx related test cases + Add abort_tx call in Producer close * Fix flaky AIO Producer test case * Fix for serializing concurrent calls from a Async callback * Minor improvements to tests * Fix styling * Fix flaky test case * Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant * Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant * Revert "Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant" This reverts commit 42b40cc. * Revert "Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant" This reverts commit f75dba3. * Add TODO * [NOGIL] Fix post-rebase gap: move rules test from test_proto_serdes.py to test_proto_serdes_rules.py * [NOGIL] Fix post-rebase gap: Fix misplaced CEL test and missing free-threaded collect_ignore entries * [NOGIL] Fix post-rebase gap: Initialize queue and add exit call in Consumer_consume * [NOGIL] Guard AdminClient against concurrent close() vs method-call races (#2317) (#2346) * Add a serializing Consumer reentrancy gate for free-threading support Introduces gate_owner/gate_depth in Consumer.c so concurrent, cross-caller access to a single Consumer/AIOConsumer instance waits for the current caller to finish rather than being undefined behavior, since librdkafka's consumer is not thread-safe; legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it) are still admitted immediately. AIOConsumer identity is tracked via a ContextVar since the owning logical caller can move across ThreadPoolExecutor worker threads. Includes unit and integration test coverage for both the sync Consumer and AIOConsumer. * Trigger CLA check * Add Integration test coverage for AIO Producer * Add Tx related test cases + Add abort_tx call in Producer close * Fix styling * Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant * Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant * Trigger CLA check * Trigger CLA check * Add a serializing Consumer reentrancy gate for free-threading support Introduces gate_owner/gate_depth in Consumer.c so concurrent, cross-caller access to a single Consumer/AIOConsumer instance waits for the current caller to finish rather than being undefined behavior, since librdkafka's consumer is not thread-safe; legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it) are still admitted immediately. AIOConsumer identity is tracked via a ContextVar since the owning logical caller can move across ThreadPoolExecutor worker threads. Includes unit and integration test coverage for both the sync Consumer and AIOConsumer. * Style fixes * Addressed comments * Trigger CLA check * Address comments * Add Tx related test cases + Add abort_tx call in Producer close * Fix for serializing concurrent calls from a Async callback * Fix one test case * Fix styling * [NOGIL] Guard AdminClient against concurrent close()-vs-method-call races * Add handle check in common APIs and refactor overall handle logic * Remove conflict marker * Remove another conflict marker * Rename handle functions * Add more tests * Fix flaky test * Address comments * Fix corrupted _common.py * Fix other corrupted files * Fix flaky test case * [NOGIL] Fix borrowed-reference/stale-count races (#2334) (#2348) * Add a serializing Consumer reentrancy gate for free-threading support Introduces gate_owner/gate_depth in Consumer.c so concurrent, cross-caller access to a single Consumer/AIOConsumer instance waits for the current caller to finish rather than being undefined behavior, since librdkafka's consumer is not thread-safe; legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it) are still admitted immediately. AIOConsumer identity is tracked via a ContextVar since the owning logical caller can move across ThreadPoolExecutor worker threads. Includes unit and integration test coverage for both the sync Consumer and AIOConsumer. * Trigger CLA check * Add Integration test coverage for AIO Producer * Add Tx related test cases + Add abort_tx call in Producer close * Fix styling * Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant * Serialize AIOConsumer calls that outlive their callback invocation instead of admitting them as re-entrant * Trigger CLA check * Trigger CLA check * Add a serializing Consumer reentrancy gate for free-threading support Introduces gate_owner/gate_depth in Consumer.c so concurrent, cross-caller access to a single Consumer/AIOConsumer instance waits for the current caller to finish rather than being undefined behavior, since librdkafka's consumer is not thread-safe; legitimate re-entrant calls (e.g. a rebalance/commit callback calling back into the Consumer that triggered it) are still admitted immediately. AIOConsumer identity is tracked via a ContextVar since the owning logical caller can move across ThreadPoolExecutor worker threads. Includes unit and integration test coverage for both the sync Consumer and AIOConsumer. * Style fixes * Addressed comments * Trigger CLA check * Address comments * Add Tx related test cases + Add abort_tx call in Producer close * Fix for serializing concurrent calls from a Async callback * Fix one test case * Fix styling * [NOGIL] Guard AdminClient against concurrent close()-vs-method-call races * Add handle check in common APIs and refactor overall handle logic * Remove conflict marker * Remove another conflict marker * Rename handle functions * Add more tests * Fix flaky test * Address comments * Fix corrupted _common.py * Fix other corrupted files * Fix flaky test case * [NOGIL] Fix borrowed-reference/stale-count races in Admin/Producer/Consumer parse loop * Fix list mutation in Admin API and a flaky test case * Fix data race on Handle.rk in close()/__exit__()/__len__() by making liveness checks atomic * Fix flaky test * Address comments * [NOGIL] Make Message class methods thread safe (#2350) * Make Message class methods thread safe * Add thread safety for eq and len functions, ref counting test cases * Use getters instead of direct access in Consumer methods * Fix mem leak * Use a single object-level lock for Message, snapshot via Message_copy, and fix __eq__ field comparison (treat unset/None alike, no crash)

Ojasva Jain (ojasvajain) requested review from a team and Matthew Seal (MSeal) as code owners July 21, 2026 13:52

Ojasva Jain (ojasvajain) self-assigned this Jul 21, 2026

Ojasva Jain (ojasvajain) added the nogil label Jul 21, 2026

airlock-confluentinc Bot force-pushed the dev_producer_no_gil branch from f1bd331 to 4c936db Compare July 23, 2026 11:37

airlock-confluentinc Bot force-pushed the dev_thread_free_support branch from 386d71a to 39740bd Compare July 23, 2026 11:39

airlock-confluentinc Bot force-pushed the dev_producer_no_gil branch from 4c936db to b9118ac Compare July 24, 2026 07:45

Kaushik Raina (k-raina) requested changes Jul 29, 2026

View reviewed changes

airlock-confluentinc Bot force-pushed the dev_thread_free_support branch from 149a22f to 448f9f6 Compare July 30, 2026 10:55

airlock-confluentinc Bot force-pushed the dev_producer_no_gil branch 2 times, most recently from b0425e3 to 7d63d3b Compare July 31, 2026 08:23

Kaushik Raina (k-raina) requested changes Aug 3, 2026

View reviewed changes

Kaushik Raina (k-raina) requested changes Aug 13, 2026

View reviewed changes

Ojasva Jain (ojasvajain) commented Aug 13, 2026

View reviewed changes

Ojasva Jain (ojasvajain) added 10 commits August 14, 2026 13:01

Replace pytest-forked with subprocess-based test isolation

0d6a9eb

Fix subprocess_isolated import and flaky close() timing assertion

75d9097

Rename tests/parallel to tests/concurrency and add integration tests …

a7453cc

…for Producer close()/transaction races

Clarified comment

48514fb

Fix close() losers to wait for winner instead of returning False early

dd0bcd5

Fix CI flakiness in test_close_races_close_losers_wait_for_slow_winner

d6a9c08

Move signal/slow-winner close() race tests to integration suite

c513c9c

Allow reentrant Producer calls from close()'s own delivery callback

4c750a2

airlock-confluentinc Bot force-pushed the dev_producer_no_gil branch from 015da26 to 4c750a2 Compare August 14, 2026 07:33

airlock-confluentinc Bot force-pushed the dev_thread_free_support branch from f11742e to 0eeb566 Compare August 14, 2026 07:34

Kaushik Raina (k-raina) requested changes Aug 17, 2026

View reviewed changes

Update docstring of test case

081b41e

Kaushik Raina (k-raina) requested changes Aug 18, 2026

View reviewed changes

Harden race checks

1c977e6

Kaushik Raina (k-raina) approved these changes Aug 18, 2026

View reviewed changes

Ojasva Jain (ojasvajain) merged commit f2f972e into dev_thread_free_support Aug 18, 2026
3 of 4 checks passed

Ojasva Jain (ojasvajain) deleted the dev_producer_no_gil branch August 18, 2026 13:18



		def test_close_waits_for_in_flight_call():
		"""close() blocks until an in-flight poll() call finishes."""

	cleanup:
	Handle_exit_rk_use(self);

	/* Cleanup resources */
	if (rkt)
	rd_kafka_topic_destroy(rkt);
	if (rkmessages)
	free(rkmessages);
	if (msgstates)
	free(msgstates);

	if (PyErr_Occurred())
	return NULL;

	return cfl_PyInt_FromInt(good);

		###############################################################################


		def test_close_completes_quickly_with_indefinite_poll_in_progress():

		assert any(all_results), f"iteration {i}: expected at least one close() call to return True, got: {all_results}"


		###############################################################################

	static int Producer_init(PyObject selfobj, PyObject args, PyObject *kwargs) {
	Handle self = (Handle )selfobj;
	char errstr[256];
	rd_kafka_conf_t *conf;

	if (self->rk) {
	PyErr_SetString(PyExc_RuntimeError,
	"Producer already __init__:ialized");
	return -1;
	}

	self->type = RD_KAFKA_PRODUCER;


		producer.close()

		def test_delivery_callback_calling_poll_does_not_crash_or_hang(self, kafka_cluster):

Conversation

Ojasva Jain (ojasvajain) commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Checklist

References

Test & Review

Open questions / Follow-ups

Uh oh!

confluent-cla-assistant Bot commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Kaushik Raina (k-raina) left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ojasva Jain (ojasvajain) Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Kaushik Raina (k-raina) Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ojasva Jain (ojasvajain) commented Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Kaushik Raina (k-raina) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Kaushik Raina (k-raina) Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Kaushik Raina (k-raina) Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Ojasva Jain (ojasvajain) commented Jul 21, 2026 •

edited

Loading

confluent-cla-assistant Bot commented Jul 21, 2026 •

edited

Loading

Kaushik Raina (k-raina) left a comment •

edited

Loading

Ojasva Jain (ojasvajain) Jul 29, 2026 •

edited

Loading

Kaushik Raina (k-raina) Aug 3, 2026 •

edited

Loading

Ojasva Jain (ojasvajain) commented Jul 29, 2026 •

edited

Loading

Kaushik Raina (k-raina) Aug 3, 2026 •

edited

Loading

Kaushik Raina (k-raina) Aug 4, 2026 •

edited

Loading

Ojasva Jain (ojasvajain) Aug 4, 2026 •

edited

Loading

Ojasva Jain (ojasvajain) Aug 6, 2026 •

edited

Loading

Ojasva Jain (ojasvajain) Aug 13, 2026 •

edited

Loading