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()