feat: Add error callback in the ProducerContext

Author: rhishikeshj

src/producer/base_producer.rs

@@ -70,6 +70,7 @@ use crate::topic_partition_list::TopicPartitionList;
 use crate::util::{Deadline, IntoOpaque, NativePtr, Timeout};
 
 pub use crate::message::DeliveryResult;
+use crate::ClientContext;
 
 use super::NoCustomPartitioner;
 
@@ -408,11 +409,17 @@ where
 
     fn handle_error_event(&self, event: NativePtr<RDKafkaEvent>) {
         let rdkafka_err = unsafe { rdsys::rd_kafka_event_error(event.ptr()) };
-        let error = KafkaError::Global(rdkafka_err.into());
+        let rdkafka_err_is_fatal = unsafe { rdsys::rd_kafka_event_error_is_fatal(event.ptr()) };
+        let error = if rdkafka_err_is_fatal != 0 {
+            KafkaError::MessageProductionFatal(rdkafka_err.into())
+        } else {
+            KafkaError::MessageProduction(rdkafka_err.into())
+        };
         let reason = unsafe {
             CStr::from_ptr(rdsys::rd_kafka_event_error_string(event.ptr())).to_string_lossy()
         };
-        self.context().error(error, reason.trim());
+        <C as ClientContext>::error(self.context(), error.clone(), reason.trim());
+        <C as ProducerContext<Part>>::error_callback(self.context(), error, reason.trim());
     }
 
     /// Returns a pointer to the native Kafka client.

src/producer/future_producer.rs

@@ -201,6 +201,10 @@ where
         };
         let _ = tx.send(owned_delivery_result); // TODO: handle error
     }
+
+    fn error_callback(&self, error: KafkaError, reason: &str) {
+        C::error(&self.wrapped_context, error, reason);
+    }
 }
 
 /// A producer that returns a [`Future`] for every message being produced.
@@ -450,6 +454,10 @@ mod tests {
         fn delivery(&self, _: &DeliveryResult<'_>, _: Self::DeliveryOpaque) {
             unimplemented!()
         }
+
+        fn error_callback(&self, _: KafkaError, _: &str) {
+            unimplemented!()
+        }
     }
 
     // Verify that the future producer is clone, according to documentation.

src/producer/mod.rs

@@ -163,7 +163,7 @@ use std::sync::Arc;
 
 use crate::client::{Client, ClientContext};
 use crate::consumer::ConsumerGroupMetadata;
-use crate::error::KafkaResult;
+use crate::error::{KafkaError, KafkaResult};
 use crate::topic_partition_list::TopicPartitionList;
 use crate::util::{IntoOpaque, Timeout};
 
@@ -201,6 +201,9 @@ pub trait ProducerContext<Part: Partitioner = NoCustomPartitioner>: ClientContex
     /// when calling send.
     fn delivery(&self, delivery_result: &DeliveryResult<'_>, delivery_opaque: Self::DeliveryOpaque);
 
+    /// Error callback
+    fn error_callback(&self, error: KafkaError, reason: &str);
+
     /// This method is called when creating producer in order to optionally register custom partitioner.
     /// If custom partitioner is not used then `partitioner` configuration property is used (or its default).
     ///
@@ -264,6 +267,8 @@ impl ProducerContext<NoCustomPartitioner> for DefaultProducerContext {
     type DeliveryOpaque = ();
 
     fn delivery(&self, _: &DeliveryResult<'_>, _: Self::DeliveryOpaque) {}
+
+    fn error_callback(&self, _: KafkaError, _: &str) {}
 }
 
 /// Common trait for all producers.

tests/test_low_producers.rs

@@ -42,6 +42,10 @@ impl ProducerContext for PrintingContext {
     fn delivery(&self, delivery_result: &DeliveryResult, delivery_opaque: Self::DeliveryOpaque) {
         println!("Delivery: {:?} {:?}", delivery_result, delivery_opaque);
     }
+
+    fn error_callback(&self, error: KafkaError, reason: &str) {
+        println!("Error: {reason:?} {error:?}");
+    }
 }
 
 type TestProducerDeliveryResult = (OwnedMessage, Option<KafkaError>, usize);
@@ -100,6 +104,10 @@ impl<Part: Partitioner + Send + Sync> ProducerContext<Part> for CollectingContex
             Some(p) => Some(p),
         }
     }
+
+    fn error_callback(&self, _: KafkaError, _: &str) {
+        unimplemented!()
+    }
 }
 
 // Partitioner sending all messages to single, defined partition.
@@ -340,6 +348,10 @@ impl ProducerContext for HeaderCheckContext {
         }
         (*self.ids.lock().unwrap()).insert(message_id);
     }
+
+    fn error_callback(&self, _: KafkaError, _: &str) {
+        unimplemented!()
+    }
 }
 
 #[test]
@@ -429,6 +441,10 @@ fn test_base_producer_opaque_arc() -> Result<(), Box<dyn Error>> {
             let mut shared_count = opaque.lock().unwrap();
             *shared_count += 1;
         }
+
+        fn error_callback(&self, _: KafkaError, _: &str) {
+            unimplemented!()
+        }
     }
 
     let shared_count = Arc::new(Mutex::new(0));