fix: Kafka source/sink - pass client certs for mTLS when insecureSkipVerify is false (#3059)

Author: BulkBeing

rust/numaflow-core/src/config/components.rs

@@ -195,59 +195,50 @@ fn parse_kafka_auth_config(
     };
 
     let tls = if let Some(tls_config) = tls_config {
-        let tls_skip_verify = tls_config.insecure_skip_verify.unwrap_or(false);
-        if tls_skip_verify {
-            Some(numaflow_kafka::TlsConfig {
-                insecure_skip_verify: true,
-                ca_cert: None,
-                client_auth: None,
+        let ca_cert = tls_config
+            .ca_cert_secret
+            .map(|ca_cert_secret| {
+                match get_secret_from_volume(&ca_cert_secret.name, &ca_cert_secret.key) {
+                    Ok(secret) => Ok(secret),
+                    Err(e) => Err(Error::Config(format!(
+                        "Failed to get CA cert secret: {e:?}"
+                    ))),
+                }
             })
-        } else {
-            let ca_cert = tls_config
-                .ca_cert_secret
-                .map(|ca_cert_secret| {
-                    match get_secret_from_volume(&ca_cert_secret.name, &ca_cert_secret.key) {
-                        Ok(secret) => Ok(secret),
-                        Err(e) => Err(Error::Config(format!(
-                            "Failed to get CA cert secret: {e:?}"
-                        ))),
-                    }
-                })
-                .transpose()?;
+            .transpose()?;
 
-            let tls_client_auth_certs = match tls_config.cert_secret {
-                Some(client_cert_secret) => {
-                    let client_cert =
-                        get_secret_from_volume(&client_cert_secret.name, &client_cert_secret.key)
-                            .map_err(|e| {
+        let tls_client_auth_certs = match tls_config.cert_secret {
+            Some(client_cert_secret) => {
+                let client_cert =
+                    get_secret_from_volume(&client_cert_secret.name, &client_cert_secret.key)
+                        .map_err(|e| {
                             Error::Config(format!("Failed to get client cert secret: {e:?}"))
                         })?;
 
-                    let Some(private_key_secret) = tls_config.key_secret else {
-                        return Err(Error::Config("Client cert is specified for TLS authentication, but private key is not specified".into()));
-                    };
+                let Some(private_key_secret) = tls_config.key_secret else {
+                    return Err(Error::Config("Client cert is specified for TLS authentication, but private key is not specified".into()));
+                };
 
-                    let client_cert_private_key =
-                        get_secret_from_volume(&private_key_secret.name, &private_key_secret.key)
-                            .map_err(|e| {
+                let client_cert_private_key =
+                    get_secret_from_volume(&private_key_secret.name, &private_key_secret.key)
+                        .map_err(|e| {
                             Error::Config(format!(
                                 "Failed to get client cert private key secret: {e:?}"
                             ))
                         })?;
-                    Some(numaflow_kafka::TlsClientAuthCerts {
-                        client_cert,
-                        client_cert_private_key,
-                    })
-                }
-                None => None,
-            };
+                Some(numaflow_kafka::TlsClientAuthCerts {
+                    client_cert,
+                    client_cert_private_key,
+                })
+            }
+            None => None,
+        };
 
-            Some(numaflow_kafka::TlsConfig {
-                insecure_skip_verify: tls_config.insecure_skip_verify.unwrap_or(false),
-                ca_cert,
-                client_auth: tls_client_auth_certs,
-            })
-        }
+        Some(numaflow_kafka::TlsConfig {
+            insecure_skip_verify: tls_config.insecure_skip_verify.unwrap_or(false),
+            ca_cert,
+            client_auth: tls_client_auth_certs,
+        })
     } else {
         None
     };