Add wrapper for depreciated API

Author: k-raina

src/confluent_kafka/src/Admin.c

@@ -5082,7 +5082,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(&exctype, &exc, &trace);
                         Py_XDECREF(trace);
                 }
 

src/confluent_kafka/src/Consumer.c

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

src/confluent_kafka/src/Producer.c

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

src/confluent_kafka/src/confluent_kafka.c

@@ -1756,7 +1756,7 @@ static void error_cb (rd_kafka_t *rk, int err, const char *reason, void *opaque)
 		Py_DECREF(result);
 	else {
 
-        PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
+        CallState_fetch_exception(cs);
         crash:
 		CallState_crash(cs);
 		rd_kafka_yield(h->rk);
@@ -1811,7 +1811,7 @@ static void throttle_cb (rd_kafka_t *rk, const char *broker_name, int32_t broker
                 Py_DECREF(result);
                 goto done;
         } else {
-                PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
+                CallState_fetch_exception(cs);
         }
 
         /**
@@ -1843,7 +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_fetch_exception(cs);
 		CallState_crash(cs);
 		rd_kafka_yield(h->rk);
 	}
@@ -1879,7 +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_fetch_exception(cs);
                 CallState_crash(cs);
                 rd_kafka_yield(h->rk);
         }
@@ -2607,10 +2607,7 @@ int CallState_end (Handle *h, CallState *cs) {
 	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;
+			CallState_restore_exception(cs);
 		}
 		return 0;
 	}

src/confluent_kafka/src/confluent_kafka.h

@@ -282,52 +282,60 @@ typedef struct {
 
 /**
  * @brief Compatibility layer for Python exception handling API changes.
- * PyErr_Fetch was deprecated in Python 3.12 in favor of PyErr_GetRaisedException.
+ * PyErr_Fetch/PyErr_Restore were deprecated in Python 3.12 in favor of 
+ * PyErr_GetRaisedException/PyErr_SetRaisedException.
  */
-#if PY_VERSION_HEX >= 0x030c0000
-/* Python 3.12+ - use new API */
+
+/* General-purpose compatibility wrappers */
 static inline void
-CallState_fetch_exception(CallState *cs) {
+cfl_exception_fetch(PyObject **exc_type, PyObject **exc_value, PyObject **exc_traceback) {
+#if PY_VERSION_HEX >= 0x030c0000
+    /* Python 3.12+ - use new API */
     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);
+        *exc_type = (PyObject *)Py_TYPE(exc);
+        Py_INCREF(*exc_type);
+        *exc_value = exc;
+        *exc_traceback = PyException_GetTraceback(exc);
     } else {
-        cs->exception_type = NULL;
-        cs->exception_value = NULL;
-        cs->exception_traceback = NULL;
+        *exc_type = *exc_value = *exc_traceback = NULL;
     }
+#else
+    /* Python < 3.12 - use legacy API */
+    PyErr_Fetch(exc_type, exc_value, exc_traceback);
+#endif
 }
 
 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;
+cfl_exception_restore(PyObject *exc_type, PyObject *exc_value, PyObject *exc_traceback) {
+#if PY_VERSION_HEX >= 0x030c0000
+    /* Python 3.12+ - use new API */
+    if (exc_value) {
+        PyErr_SetRaisedException(exc_value);
+        Py_XDECREF(exc_type);
+        Py_XDECREF(exc_traceback);
     }
-}
 #else
-/* Python < 3.12 - use legacy API */
+    /* Python < 3.12 - use legacy API */
+    PyErr_Restore(exc_type, exc_value, exc_traceback);
+#endif
+}
+
+/* CallState-specific convenience wrappers */
 static inline void
 CallState_fetch_exception(CallState *cs) {
-    PyErr_Fetch(&cs->exception_type, &cs->exception_value, &cs->exception_traceback);
+    cfl_exception_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;
+    if (cs->exception_type) {
+        cfl_exception_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