[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.

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)

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

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 from 85f31c7 to 418c19c Compare July 30, 2026 11:24

Ojasva Jain (ojasvajain) added 6 commits July 31, 2026 13:53

Replace pytest-forked with subprocess-based test isolation

588be11

Fix subprocess_isolated import and flaky close() timing assertion

27e14d5

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

b42644d

…for Producer close()/transaction races

Clarified comment

9058328

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

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

View reviewed changes

Ojasva Jain (ojasvajain) added 3 commits August 4, 2026 18:12

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

b17700d

Fix CI flakiness in test_close_races_close_losers_wait_for_slow_winner

ef44e17

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

ba1020c



		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}"


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

Uh oh!

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

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