Address PR review feedback

- Add issue number #865 to test comments for better traceability
- Move PyErr_Fetch into else condition for consistency in confluent_kafka.c
- Add CHANGELOG entry documenting the fix for issue #865

Author: k-raina

CHANGELOG.md

@@ -4,6 +4,9 @@
 
 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/confluent_kafka.c

@@ -1810,10 +1810,10 @@ 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 {
+                PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
         }
 
-        PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
-
         /**
         * Stop callback dispatcher, return err to application
         * fall-through to unlock GIL

src/confluent_kafka/src/confluent_kafka.h

@@ -280,6 +280,55 @@ typedef struct {
 	PyObject *exception_traceback; /* Stored exception traceback */
 } CallState;
 
+/**
+ * @brief Compatibility layer for Python exception handling API changes.
+ * PyErr_Fetch was deprecated in Python 3.12 in favor of PyErr_GetRaisedException.
+ */
+#if PY_VERSION_HEX >= 0x030c0000
+/* Python 3.12+ - use new API */
+static inline void
+CallState_fetch_exception(CallState *cs) {
+    PyObject *exc = PyErr_GetRaisedException();
+    if (exc) {
+        cs->exception_type = (PyObject *)Py_TYPE(exc);
+        Py_INCREF(cs->exception_type);
+        cs->exception_value = exc;
+        cs->exception_traceback = PyException_GetTraceback(exc);
+    } else {
+        cs->exception_type = NULL;
+        cs->exception_value = NULL;
+        cs->exception_traceback = NULL;
+    }
+}
+
+static inline void
+CallState_restore_exception(CallState *cs) {
+    if (cs->exception_value) {
+        PyErr_SetRaisedException(cs->exception_value);
+        /* PyErr_SetRaisedException steals the reference, so clear our pointer */
+        cs->exception_value = NULL;
+        Py_XDECREF(cs->exception_type);
+        cs->exception_type = NULL;
+        Py_XDECREF(cs->exception_traceback);
+        cs->exception_traceback = NULL;
+    }
+}
+#else
+/* Python < 3.12 - use legacy API */
+static inline void
+CallState_fetch_exception(CallState *cs) {
+    PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
+}
+
+static inline void
+CallState_restore_exception(CallState *cs) {
+    PyErr_Restore(cs->exception_type, cs->exception_value, cs->exception_traceback);
+    cs->exception_type = NULL;
+    cs->exception_value = NULL;
+    cs->exception_traceback = NULL;
+}
+#endif
+
 /**
  * @brief Initialiase a CallState and unlock the GIL prior to a
  *        possibly blocking external call.

tests/test_Consumer.py

@@ -347,7 +347,7 @@ def error_cb_that_raises(error):
     consumer.subscribe(['test-topic'])
 
     # This should trigger the error callback due to connection failure
-    # Before fix: Would get RuntimeError + SystemError
+    # Before fix: Would get RuntimeError + SystemError (Issue #865)
     # After fix: Should only get RuntimeError (no SystemError)
     with pytest.raises(RuntimeError) as exc_info:
         consumer.consume(timeout=0.1)

tests/test_Producer.py

@@ -304,7 +304,7 @@ def delivery_cb_that_raises(err, msg):
     producer.produce('test-topic', value='test-message')
 
     # Flush to ensure delivery reports are processed
-    # Before fix: Would get RuntimeError + SystemError
+    # Before fix: Would get RuntimeError + SystemError (Issue #865)
     # After fix: Should only get RuntimeError (no SystemError)
     with pytest.raises(RuntimeError) as exc_info:
         producer.flush(timeout=2.0)  # Longer timeout to ensure delivery callback fires