diff --git a/rust-arroyo/src/backends/kafka/errors.rs b/rust-arroyo/src/backends/kafka/errors.rs index 7cdbb108..a4592851 100644 --- a/rust-arroyo/src/backends/kafka/errors.rs +++ b/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 for ConsumerError { } } } - -impl From for ProducerError { - fn from(err: KafkaError) -> Self { - ProducerError::ProducerFailure { - error: get_error_name(&err), - } - } -} diff --git a/rust-arroyo/src/backends/kafka/producer.rs b/rust-arroyo/src/backends/kafka/producer.rs index c0d3eff7..d25ea6bd 100644 --- a/rust-arroyo/src/backends/kafka/producer.rs +++ b/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( diff --git a/rust-arroyo/src/backends/mod.rs b/rust-arroyo/src/backends/mod.rs index 2a1bbd2d..56a1b058 100755 --- a/rust-arroyo/src/backends/mod.rs +++ b/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 }, }