Fix error propagation rule for Python's C API

Author: k-raina

src/confluent_kafka/src/Consumer.c

@@ -1575,6 +1575,7 @@ static void Consumer_rebalance_cb (rd_kafka_t *rk, rd_kafka_resp_err_t err,
 if (result)
 Py_DECREF(result);
 else {
+PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
 CallState_crash(cs);
 rd_kafka_yield(rk);
 }

src/confluent_kafka/src/Producer.c

@@ -163,6 +163,7 @@ static void dr_msg_cb (rd_kafka_t *rk, const rd_kafka_message_t *rkm,
 if (result)
 Py_DECREF(result);
 else {
+PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
 CallState_crash(cs);
 rd_kafka_yield(rk);
 }

src/confluent_kafka/src/confluent_kafka.c

@@ -1755,6 +1755,8 @@ static void error_cb (rd_kafka_t *rk, int err, const char *reason, void *opaque)
 if (result)
 Py_DECREF(result);
 else {
+ 
+ PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
  crash:
 CallState_crash(cs);
 rd_kafka_yield(h->rk);
@@ -1810,6 +1812,8 @@ static void throttle_cb (rd_kafka_t *rk, const char *broker_name, int32_t broker
  goto done;
  }
 
+ PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
+
  /**
  * Stop callback dispatcher, return err to application
  * fall-through to unlock GIL
@@ -1839,6 +1843,7 @@ static int stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque) {
 if (result)
 Py_DECREF(result);
 else {
+PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
 CallState_crash(cs);
 rd_kafka_yield(h->rk);
 }
@@ -1874,6 +1879,7 @@ static void log_cb (const rd_kafka_t *rk, int level,
  if (result)
  Py_DECREF(result);
  else {
+ PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
  CallState_crash(cs);
  rd_kafka_yield(h->rk);
  }
@@ -2572,6 +2578,9 @@ void CallState_begin (Handle *h, CallState *cs) {
 cs->thread_state = PyEval_SaveThread();
 assert(cs->thread_state != NULL);
 cs->crashed = 0;
+cs->exception_type = NULL;
+cs->exception_value = NULL;
+cs->exception_traceback = NULL;
 #ifdef WITH_PY_TSS
  PyThread_tss_set(&h->tlskey, cs);
 #else
@@ -2592,8 +2601,19 @@ 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_type) {
+PyErr_Restore(cs->exception_type, cs->exception_value, cs->exception_traceback);
+cs->exception_type = NULL;
+cs->exception_value = NULL;
+cs->exception_traceback = NULL;
+}
 return 0;
+}
 
 return 1;
 }

src/confluent_kafka/src/confluent_kafka.h

@@ -275,6 +275,9 @@ int Handle_traverse (Handle *h, visitproc visit, void *arg);
 typedef struct {
 PyThreadState *thread_state;
 int crashed; /* Callback crashed */
+PyObject *exception_type; /* Stored exception type */
+PyObject *exception_value; /* Stored exception value */
+PyObject *exception_traceback; /* Stored exception traceback */
 } CallState;
 
 /**

tests/test_Consumer.py

@@ -324,3 +324,38 @@ def test_consumer_without_groupid():
  with pytest.raises(ValueError) as ex:
  TestConsumer({'bootstrap.servers': "mybroker:9092"})
  assert ex.match('group.id must be set')
+
+
+def test_callback_exception_no_system_error():
+
+ exception_raised = []
+ 
+ def error_cb_that_raises(error):
+ """Error callback that raises an exception"""
+ exception_raised.append(error)
+ raise RuntimeError("Test exception from error_cb")
+ 
+ # Create consumer with error callback that raises exception
+ consumer = TestConsumer({
+ 'group.id': 'test-callback-systemerror-fix',
+ 'bootstrap.servers': 'nonexistent-broker:9092', # Will trigger error
+ 'socket.timeout.ms': 100,
+ 'session.timeout.ms': 1000,
+ 'error_cb': error_cb_that_raises
+ })
+ 
+ consumer.subscribe(['test-topic'])
+ 
+ # This should trigger the error callback due to connection failure
+ # Before fix: Would get RuntimeError + SystemError
+ # After fix: Should only get RuntimeError (no SystemError)
+ with pytest.raises(RuntimeError) as exc_info:
+ consumer.consume(timeout=0.1)
+ 
+ # Verify we got the expected exception message
+ assert "Test exception from error_cb" in str(exc_info.value)
+ 
+ # Verify the error callback was actually called
+ assert len(exception_raised) > 0
+ 
+ consumer.close()

tests/test_Producer.py

@@ -283,3 +283,36 @@ def test_producer_bool_value():
 
  p = Producer({})
  assert bool(p)
+
+
+def test_callback_exception_no_system_error():
+ delivery_reports = []
+ 
+ def delivery_cb_that_raises(err, msg):
+ """Delivery report callback that raises an exception"""
+ delivery_reports.append((err, msg))
+ raise RuntimeError("Test exception from delivery_cb")
+ 
+ producer = Producer({
+ 'bootstrap.servers': 'nonexistent-broker:9092', # Will cause delivery failures
+ 'socket.timeout.ms': 100,
+ 'message.timeout.ms': 1000,
+ 'on_delivery': delivery_cb_that_raises
+ })
+ 
+ # Produce a message - this will trigger delivery report callback when it fails
+ producer.produce('test-topic', value='test-message')
+ 
+ # Flush to ensure delivery reports are processed
+ # Before fix: Would get RuntimeError + SystemError 
+ # After fix: Should only get RuntimeError (no SystemError)
+ with pytest.raises(RuntimeError) as exc_info:
+ producer.flush(timeout=1.0)
+ 
+ # Verify we got an exception from our callback
+ assert "Test exception from delivery_cb" in str(exc_info.value)
+ 
+ # Verify the delivery callback was actually called
+ assert len(delivery_reports) > 0
+ 
+ producer.close()