feat: Add error callback in the ConsumerContext

Author: rhishikeshj

src/consumer/base_consumer.rs

@@ -241,16 +241,22 @@ where
     fn handle_error_event(&self, event: NativePtr<RDKafkaEvent>) -> Option<KafkaError> {
         let rdkafka_err = unsafe { rdsys::rd_kafka_event_error(event.ptr()) };
         if rdkafka_err.is_error() {
-            if rdkafka_err == rdsys::rd_kafka_resp_err_t::RD_KAFKA_RESP_ERR__PARTITION_EOF {
-                let tp_ptr = unsafe { rdsys::rd_kafka_event_topic_partition(event.ptr()) };
-                let partition = unsafe { (*tp_ptr).partition };
-                unsafe { rdsys::rd_kafka_topic_partition_destroy(tp_ptr) };
-                Some(KafkaError::PartitionEOF(partition))
-            } else if unsafe { rdsys::rd_kafka_event_error_is_fatal(event.ptr()) } != 0 {
-                Some(KafkaError::MessageConsumptionFatal(rdkafka_err.into()))
-            } else {
-                Some(KafkaError::MessageConsumption(rdkafka_err.into()))
-            }
+            let error =
+                if rdkafka_err == rdsys::rd_kafka_resp_err_t::RD_KAFKA_RESP_ERR__PARTITION_EOF {
+                    let tp_ptr = unsafe { rdsys::rd_kafka_event_topic_partition(event.ptr()) };
+                    let partition = unsafe { (*tp_ptr).partition };
+                    unsafe { rdsys::rd_kafka_topic_partition_destroy(tp_ptr) };
+                    KafkaError::PartitionEOF(partition)
+                } else if unsafe { rdsys::rd_kafka_event_error_is_fatal(event.ptr()) } != 0 {
+                    KafkaError::MessageConsumptionFatal(rdkafka_err.into())
+                } else {
+                    KafkaError::MessageConsumption(rdkafka_err.into())
+                };
+            let reason = unsafe {
+                CStr::from_ptr(rdsys::rd_kafka_event_error_string(event.ptr())).to_string_lossy()
+            };
+            self.context().error_callback(error.clone(), reason.trim());
+            Some(error)
         } else {
             None
         }

src/consumer/mod.rs

@@ -130,6 +130,10 @@ pub trait ConsumerContext: ClientContext + Sized {
     fn main_queue_min_poll_interval(&self) -> Timeout {
         Timeout::After(Duration::from_secs(1))
     }
+
+    /// Error callback
+    #[allow(unused_variables)]
+    fn error_callback(&self, error: KafkaError, reason: &str) {}
 }
 
 /// An inert [`ConsumerContext`] that can be used when no customizations are