fix: kafka security protocol

Author: Devdutt Shenoi

src/cli.rs

@@ -318,33 +318,32 @@ pub struct Options {
 
     // Kafka configuration (conditionally compiled)
     #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
-    #[arg(
-        long,env = "P_KAFKA_TOPICS", help = "Kafka topics to subscribe to")]
+    #[arg(long, env = "P_KAFKA_TOPICS", help = "Kafka topics to subscribe to")]
     pub kafka_topics: Option<String>,
 
     #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
-    #[arg(
-        long,env = "P_KAFKA_HOST", help = "Address and port for Kafka server")]
+    #[arg(long, env = "P_KAFKA_HOST", help = "Address and port for Kafka server")]
     pub kafka_host: Option<String>,
 
     #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
-    #[arg(
-        long,env = "P_KAFKA_GROUP", help = "Kafka group")]
+    #[arg(long, env = "P_KAFKA_GROUP", help = "Kafka group")]
     pub kafka_group: Option<String>,
 
     #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
-    #[arg(
-        long,env = "P_KAFKA_CLIENT_ID", help = "Kafka client id")]
+    #[arg(long, env = "P_KAFKA_CLIENT_ID", help = "Kafka client id")]
     pub kafka_client_id: Option<String>,
 
     #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
     #[arg(
-        long,env = "P_KAFKA_SECURITY_PROTOCOL", help = "Kafka security protocol")]
+        long,
+        env = "P_KAFKA_SECURITY_PROTOCOL",
+        value_parser = validation::kafka_security_protocol,
+        help = "Kafka security protocol"
+    )]
     pub kafka_security_protocol: Option<KafkaSslProtocol>,
 
     #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
-    #[arg(
-        long,env = "P_KAFKA_PARTITIONS", help = "Kafka partitions")]
+    #[arg(long, env = "P_KAFKA_PARTITIONS", help = "Kafka partitions")]
     pub kafka_partitions: Option<String>,
 
     // Audit logging

src/option.rs

@@ -28,6 +28,9 @@ use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 use std::sync::Arc;
 
+#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+use crate::kafka::KafkaSslProtocol;
+
 pub const JOIN_COMMUNITY: &str =
     "Join us on Parseable Slack community for questions : https://logg.ing/community";
 pub static CONFIG: Lazy<Arc<Config>> = Lazy::new(|| Arc::new(Config::new()));
@@ -243,6 +246,17 @@ pub mod validation {
         url::Url::parse(s).map_err(|_| "Invalid URL provided".to_string())
     }
 
+    #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
+    pub fn kafka_security_protocol(s: &str) -> Result<KafkaSslProtocol, String> {
+        match s {
+            "plaintext" => Ok(KafkaSslProtocol::Plaintext),
+            "ssl" => Ok(KafkaSslProtocol::Ssl),
+            "sasl_plaintext" => Ok(KafkaSslProtocol::SaslPlaintext),
+            "sasl_ssl" => Ok(KafkaSslProtocol::SaslSsl),
+            _ => Err("Invalid Kafka Security Protocol provided".to_string()),
+        }
+    }
+
     pub fn mode(s: &str) -> Result<Mode, String> {
         match s {
             "query" => Ok(Mode::Query),