fix(producer): return original Kafka errors from synchronous producer

rust-arroyo/src/backends/kafka/errors.rs

@@ -1,7 +1,6 @@
 use rdkafka::error::{KafkaError, RDKafkaErrorCode};
 
 use crate::backends::ConsumerError;
-use crate::backends::ProducerError;
 
 /// Returns a string representation of the KafkaError variant name and any embedded RDKafkaErrorCode
 pub fn get_error_name(error: &KafkaError) -> String {
@@ -53,11 +52,3 @@ impl From<KafkaError> for ConsumerError {
         }
     }
 }
-
-impl From<KafkaError> for ProducerError {
-    fn from(err: KafkaError) -> Self {
-        ProducerError::ProducerFailure {
-            error: get_error_name(&err),
-        }
-    }
-}

rust-arroyo/src/backends/kafka/producer.rs

@@ -264,13 +264,23 @@ mod tests {
     use super::{AsyncKafkaProducer, KafkaProducer, ProducerContext};
     use crate::backends::kafka::config::KafkaConfig;
     use crate::backends::kafka::types::KafkaPayload;
-    use crate::backends::{AsyncProducer, Producer};
+    use crate::backends::{AsyncProducer, Producer, ProducerError};
     use crate::types::{Topic, TopicOrPartition};
     use rdkafka::client::ClientContext;
-    use rdkafka::error::KafkaError;
+    use rdkafka::error::{KafkaError, RDKafkaErrorCode};
     use rdkafka::statistics::{Broker, Statistics, Window};
     use std::collections::HashMap;
 
+    fn queue_full_configuration() -> KafkaConfig {
+        KafkaConfig::new_producer_config(
+            Vec::new(),
+            Some(HashMap::from([
+                ("queue.buffering.max.messages".to_string(), "1".to_string()),
+                ("message.timeout.ms".to_string(), "5000".to_string()),
+            ])),
+        )
+    }
+
     fn create_test_statistics_with_all_metrics() -> Statistics {
         let mut brokers = HashMap::new();
         brokers.insert(
@@ -438,6 +448,31 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_sync_enqueue_error_retains_queue_full_error() {
+        let producer = KafkaProducer::new(queue_full_configuration());
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
+        let destination = TopicOrPartition::Topic(Topic::new("test"));
+
+        let first_result = producer.produce(
+            &destination,
+            KafkaPayload::new(None, None, Some(b"first".to_vec())),
+        );
+        assert!(first_result.is_ok());
+
+        let second_result = producer.produce(
+            &destination,
+            KafkaPayload::new(None, None, Some(b"second".to_vec())),
+        );
+        assert!(matches!(
+            second_result,
+            Err(ProducerError::Kafka(KafkaError::MessageProduction(
+                RDKafkaErrorCode::QueueFull
+            )))
+        ));
+    }
+
     #[test]
     fn test_invalid_producer_configuration_returns_error() {
         let configuration = KafkaConfig::new_producer_config(

rust-arroyo/src/backends/mod.rs

@@ -1,4 +1,5 @@
 use super::types::{BrokerMessage, Partition, TopicOrPartition};
+use rdkafka::error::KafkaError;
 use std::collections::{HashMap, HashSet};
 use std::future::Future;
 use std::pin::Pin;
@@ -43,6 +44,9 @@ pub enum ProducerError {
     #[error("The producer errored")]
     ProducerErrored,
 
+    #[error(transparent)]
+    Kafka(#[from] KafkaError),
+
     #[error("Producer errored with code")]
     ProducerFailure { error: String },
 }