Fix error propagation rule for Python's C API

CHANGELOG.md

@@ -78,6 +78,9 @@ for a complete list of changes, enhancements, fixes and upgrade considerations.
 
 v2.11.0 is a feature release with the following enhancements:
 
+### Fixes
+- Fix error propagation rule for Python's C API to prevent SystemError when callbacks raise exceptions (#865)
+
 confluent-kafka-python v2.11.0 is based on librdkafka v2.11.0, see the
 [librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.11.0)
 for a complete list of changes, enhancements, fixes and upgrade considerations.

src/confluent_kafka/src/Admin.c

@@ -4717,7 +4717,7 @@ static void Admin_background_event_cb (rd_kafka_t *rk, rd_kafka_event_t *rkev,
         PyGILState_STATE gstate;
         PyObject *error, *method, *ret;
         PyObject *result = NULL;
-        PyObject *exctype = NULL, *exc = NULL, *excargs = NULL;
+        PyObject *exc = NULL, *excargs = NULL;
 
         /* Acquire GIL */
         gstate = PyGILState_Ensure();
@@ -5093,7 +5093,7 @@ static void Admin_background_event_cb (rd_kafka_t *rk, rd_kafka_event_t *rkev,
                         PyObject *trace = NULL;
 
                         /* Fetch (and clear) currently raised exception */
-                        PyErr_Fetch(&exctype, &error, &trace);
+                        cfl_exception_fetch(&exc);
                         Py_XDECREF(trace);
                 }
 
@@ -5124,22 +5124,17 @@ static void Admin_background_event_cb (rd_kafka_t *rk, rd_kafka_event_t *rkev,
          * Pass an exception to future.set_exception().
          */
 
-        if (!exctype) {
+        if (!exc) {
                 /* No previous exception raised, use KafkaException */
-                exctype = KafkaException;
-                Py_INCREF(exctype);
-        }
-
-        /* Create a new exception based on exception type and error. */
-        excargs = PyTuple_New(1);
-        Py_INCREF(error); /* tuple's reference */
-        PyTuple_SET_ITEM(excargs, 0, error);
-        exc = ((PyTypeObject *)exctype)->tp_new(
-                (PyTypeObject *)exctype, NULL, NULL);
-        exc->ob_type->tp_init(exc, excargs, NULL);
-        Py_DECREF(excargs);
-        Py_XDECREF(exctype);
-        Py_XDECREF(error); /* from error source above */
+                excargs = PyTuple_New(1);
+                Py_INCREF(error); /* tuple's reference */
+                PyTuple_SET_ITEM(excargs, 0, error);
+                exc = ((PyTypeObject *)KafkaException)->tp_new(
+                        (PyTypeObject *)KafkaException, NULL, NULL);
+                exc->ob_type->tp_init(exc, excargs, NULL);
+                Py_DECREF(excargs);
+                Py_XDECREF(error); /* from error source above */
+        }
 
         /*
          * Call future.set_exception(exc)

src/confluent_kafka/src/Consumer.c

@@ -424,6 +424,7 @@ static void Consumer_offset_commit_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,
         if (result)
                 Py_DECREF(result);
         else {
+                CallState_fetch_exception(cs);
                 CallState_crash(cs);
                 rd_kafka_yield(rk);
         }
@@ -1586,6 +1587,7 @@ static void Consumer_rebalance_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,
 		if (result)
 			Py_DECREF(result);
 		else {
+			CallState_fetch_exception(cs);
 			CallState_crash(cs);
 			rd_kafka_yield(rk);
 		}

src/confluent_kafka/src/Producer.c

@@ -174,6 +174,7 @@ static void dr_msg_cb (rd_kafka_t *rk, const rd_kafka_message_t *rkm,
 	if (result)
 		Py_DECREF(result);
 	else {
+		CallState_fetch_exception(cs);
 		CallState_crash(cs);
 		rd_kafka_yield(rk);
 	}

src/confluent_kafka/src/confluent_kafka.c

@@ -1766,6 +1766,7 @@ static void error_cb (rd_kafka_t *rk, int err, const char *reason, void *opaque)
 	if (result)
 		Py_DECREF(result);
 	else {
+		CallState_fetch_exception(cs);
         crash:
 		CallState_crash(cs);
 		rd_kafka_yield(h->rk);
@@ -1819,6 +1820,8 @@ static void throttle_cb (rd_kafka_t *rk, const char *broker_name, int32_t broker
                 /* throttle_cb executed successfully */
                 Py_DECREF(result);
                 goto done;
+        } else {
+                CallState_fetch_exception(cs);
         }
 
         /**
@@ -1850,6 +1853,7 @@ static int stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque) {
 	if (result)
 		Py_DECREF(result);
 	else {
+		CallState_fetch_exception(cs);
 		CallState_crash(cs);
 		rd_kafka_yield(h->rk);
 	}
@@ -1885,6 +1889,7 @@ static void log_cb (const rd_kafka_t *rk, int level,
         if (result)
                 Py_DECREF(result);
         else {
+                CallState_fetch_exception(cs);
                 CallState_crash(cs);
                 rd_kafka_yield(h->rk);
         }
@@ -2583,6 +2588,7 @@ void CallState_begin (Handle *h, CallState *cs) {
 	cs->thread_state = PyEval_SaveThread();
 	assert(cs->thread_state != NULL);
 	cs->crashed = 0;
+	cs->exception_value = NULL;
 #ifdef WITH_PY_TSS
         PyThread_tss_set(&h->tlskey, cs);
 #else
@@ -2603,9 +2609,17 @@ int CallState_end (Handle *h, CallState *cs) {
 
 	PyEval_RestoreThread(cs->thread_state);
 
-	if (PyErr_CheckSignals() == -1 || cs->crashed)
+	if (PyErr_CheckSignals() == -1)
 		return 0;
 
+	if (cs->crashed) {
+		/* Restore the saved exception if we have one */
+		if (cs->exception_value) {
+			CallState_restore_exception(cs);
+		}
+		return 0;
+	}
+
 	return 1;
 }
 

src/confluent_kafka/src/confluent_kafka.h

@@ -270,8 +270,55 @@ int  Handle_traverse (Handle *h, visitproc visit, void *arg);
 typedef struct {
 	PyThreadState *thread_state;
 	int crashed;   /* Callback crashed */
+	PyObject *exception_value;   /* Stored exception value */
 } CallState;
 
+/**
+ * @brief Compatibility layer for Python exception handling API changes.
+ * PyErr_Fetch/PyErr_Restore were deprecated in Python 3.12 in favor of 
+ * PyErr_GetRaisedException/PyErr_SetRaisedException.
+ */
+
+static inline void
+cfl_exception_fetch(PyObject **exc_value) {
+#if PY_VERSION_HEX >= 0x030c0000
+    /* Python 3.12+ - use new API */
+    *exc_value = PyErr_GetRaisedException();
+#else
+    /* Python < 3.12 - use legacy API */
+    PyObject *exc_type, *exc_traceback;
+    PyErr_Fetch(&exc_type, exc_value, &exc_traceback);
+    Py_XDECREF(exc_type);
+    Py_XDECREF(exc_traceback);
+#endif
+}
+
+static inline void
+cfl_exception_restore(PyObject *exc_value) {
+#if PY_VERSION_HEX >= 0x030c0000
+    /* Python 3.12+ - use new API */
+    if (exc_value) {
+        PyErr_SetRaisedException(exc_value);
+    }
+#else
+    /* Python < 3.12 - use legacy API */
+    PyErr_SetObject(PyExceptionInstance_Class(exc_value), exc_value);
+#endif
+}
+
+static inline void
+CallState_fetch_exception(CallState *cs) {
+    cfl_exception_fetch(&cs->exception_value);
+}
+
+static inline void
+CallState_restore_exception(CallState *cs) {
+    if (cs->exception_value) {
+        cfl_exception_restore(cs->exception_value);
+        cs->exception_value = NULL;
+    }
+}
+
 /**
  * @brief Initialiase a CallState and unlock the GIL prior to a
  *        possibly blocking external call.