fix(producer): make Rust Kafka producer constructors fallible

rust-arroyo/examples/transform_and_produce.rs

@@ -49,6 +49,8 @@ async fn main() {
     impl ProcessingStrategyFactory<KafkaPayload> for ReverseStringAndProduceStrategyFactory {
         fn create(&self) -> Box<dyn ProcessingStrategy<KafkaPayload>> {
             let producer = KafkaProducer::new(self.config.clone());
+            assert!(producer.is_ok());
+            let producer = producer.unwrap();
             let topic = TopicOrPartition::Topic(self.topic);
             let reverse_string_and_produce_strategy = RunTask::new(
                 reverse_string,

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

@@ -679,6 +679,8 @@ mod tests {
         );
 
         let producer = KafkaProducer::new(producer_configuration);
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
         let payload = KafkaPayload::new(None, None, Some("asdf".as_bytes().to_vec()));
 
         producer

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

@@ -139,20 +139,19 @@ pub struct KafkaProducer {
 }
 
 impl KafkaProducer {
-    pub fn new(config: KafkaConfig) -> Self {
+    pub fn new(config: KafkaConfig) -> Result<Self, KafkaError> {
         // Extract client.id from config for metrics, default to "unknown"
         let producer_name = config
             .get_config_value("client.id")
             .cloned()
             .unwrap_or_else(|| "unknown".to_string());
         let context = ProducerContext::new(producer_name.clone());
         let config_obj: ClientConfig = config.into();
-        let threaded_producer: ThreadedProducer<_> =
-            config_obj.create_with_context(context).unwrap();
+        let threaded_producer: ThreadedProducer<_> = config_obj.create_with_context(context)?;
 
-        Self {
+        Ok(Self {
             producer: threaded_producer,
-        }
+        })
     }
 }
 
@@ -178,20 +177,20 @@ pub struct AsyncKafkaProducer {
 }
 
 impl AsyncKafkaProducer {
-    pub fn new(config: KafkaConfig) -> Self {
+    pub fn new(config: KafkaConfig) -> Result<Self, KafkaError> {
         // Extract client.id from config for metrics, default to "unknown"
         let producer_name = config
             .get_config_value("client.id")
             .cloned()
             .unwrap_or_else(|| "unknown".to_string());
         let context = ProducerContext::new(producer_name.clone());
         let config_obj: ClientConfig = config.into();
-        let future_producer: FutureProducer<_> = config_obj.create_with_context(context).unwrap();
+        let future_producer: FutureProducer<_> = config_obj.create_with_context(context)?;
 
-        Self {
+        Ok(Self {
             producer: future_producer,
             producer_name,
-        }
+        })
     }
 }
 
@@ -268,6 +267,7 @@ mod tests {
     use crate::backends::{AsyncProducer, Producer};
     use crate::types::{Topic, TopicOrPartition};
     use rdkafka::client::ClientContext;
+    use rdkafka::error::KafkaError;
     use rdkafka::statistics::{Broker, Statistics, Window};
     use std::collections::HashMap;
 
@@ -389,11 +389,13 @@ mod tests {
             KafkaConfig::new_producer_config(vec!["127.0.0.1:9092".to_string()], None);
 
         let producer = KafkaProducer::new(configuration);
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
 
         let payload = KafkaPayload::new(None, None, Some("asdf".as_bytes().to_vec()));
         producer
             .produce(&destination, payload)
-            .expect("Message produced")
+            .expect("Message produced");
     }
 
     #[tokio::test]
@@ -404,6 +406,8 @@ mod tests {
             KafkaConfig::new_producer_config(vec!["127.0.0.1:9092".to_string()], None);
 
         let producer = AsyncKafkaProducer::new(configuration);
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
 
         let payload = KafkaPayload::new(None, None, Some("asdf".as_bytes().to_vec()));
         let result = producer.produce(&destination, payload).await;
@@ -423,6 +427,8 @@ mod tests {
         );
 
         let producer = AsyncKafkaProducer::new(configuration);
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
 
         let payload = KafkaPayload::new(None, None, Some("asdf".as_bytes().to_vec()));
         let result = producer.produce(&destination, payload).await;
@@ -431,4 +437,36 @@ mod tests {
             "Message should not be produced successfully"
         );
     }
+
+    #[test]
+    fn test_invalid_producer_configuration_returns_error() {
+        let configuration = KafkaConfig::new_producer_config(
+            vec!["127.0.0.1:9092".to_string()],
+            Some(HashMap::from([(
+                "message.timeout.ms".to_string(),
+                "invalid".to_string(),
+            )])),
+        );
+
+        assert!(matches!(
+            KafkaProducer::new(configuration),
+            Err(KafkaError::ClientConfig(..))
+        ));
+    }
+
+    #[test]
+    fn test_invalid_async_producer_configuration_returns_error() {
+        let configuration = KafkaConfig::new_producer_config(
+            vec!["127.0.0.1:9092".to_string()],
+            Some(HashMap::from([(
+                "message.timeout.ms".to_string(),
+                "invalid".to_string(),
+            )])),
+        );
+
+        assert!(matches!(
+            AsyncKafkaProducer::new(configuration),
+            Err(KafkaError::ClientConfig(..))
+        ));
+    }
 }

rust-arroyo/src/processing/strategies/produce.rs

@@ -157,7 +157,9 @@ mod tests {
 
         let partition = Partition::new(Topic::new("test"), 0);
 
-        let producer: KafkaProducer = KafkaProducer::new(config);
+        let producer = KafkaProducer::new(config);
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
         let concurrency = ConcurrencyConfig::new(10);
         let mut strategy = Produce::new(
             Noop {},

rust-arroyo/src/testutils.rs

@@ -107,6 +107,8 @@ impl TestTopic {
             KafkaConfig::new_producer_config(vec![get_default_broker()], None);
 
         let producer = KafkaProducer::new(producer_configuration);
+        assert!(producer.is_ok());
+        let producer = producer.unwrap();
 
         producer
             .produce(&crate::types::TopicOrPartition::Topic(self.topic), payload)