fix: We don't actually need new callbacks!

We just need to trigger the existing one in the ClientContext

Author: rhishikeshj

src/consumer/base_consumer.rs

@@ -255,7 +255,7 @@ where
             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());
+            self.context().error(error.clone(), reason.trim());
             Some(error)
         } else {
             None

src/consumer/mod.rs

@@ -130,10 +130,6 @@ 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

src/producer/base_producer.rs

@@ -70,7 +70,6 @@ use crate::topic_partition_list::TopicPartitionList;
 use crate::util::{Deadline, IntoOpaque, NativePtr, Timeout};
 
 pub use crate::message::DeliveryResult;
-use crate::ClientContext;
 
 use super::NoCustomPartitioner;
 
@@ -418,8 +417,7 @@ where
         let reason = unsafe {
             CStr::from_ptr(rdsys::rd_kafka_event_error_string(event.ptr())).to_string_lossy()
         };
-        <C as ClientContext>::error(self.context(), error.clone(), reason.trim());
-        <C as ProducerContext<Part>>::error_callback(self.context(), error, reason.trim());
+        self.context().error(error.clone(), reason.trim());
     }
 
     /// Returns a pointer to the native Kafka client.

src/producer/future_producer.rs

@@ -201,10 +201,6 @@ 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.
@@ -454,10 +450,6 @@ 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::{KafkaError, KafkaResult};
+use crate::error::KafkaResult;
 use crate::topic_partition_list::TopicPartitionList;
 use crate::util::{IntoOpaque, Timeout};
 
@@ -201,9 +201,6 @@ 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).
     ///
@@ -267,8 +264,6 @@ 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,10 +42,6 @@ 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);
@@ -104,10 +100,6 @@ 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.
@@ -348,10 +340,6 @@ impl ProducerContext for HeaderCheckContext {
         }
         (*self.ids.lock().unwrap()).insert(message_id);
     }
-
-    fn error_callback(&self, _: KafkaError, _: &str) {
-        unimplemented!()
-    }
 }
 
 #[test]
@@ -441,10 +429,6 @@ 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));