fix(mqtt): pass client certificates to rumqttc for mTLS

Author: mr-

changelog.d/mqtt_source_mtls.fix.md

@@ -0,0 +1,3 @@
+Fixed the mqtt sink's certificate handling.
+
+authors: mr-

src/sources/mqtt/config.rs

@@ -148,7 +148,7 @@ impl MqttSourceConfig {
 
         if let Some(tls) = tls.tls() {
             let ca = tls.authorities_pem().flatten().collect();
-            let client_auth = None;
+            let client_auth = tls.identity_pem();
             let alpn = Some(vec!["mqtt".into()]);
             options.set_transport(Transport::Tls(TlsConfiguration::Simple {
                 ca,

src/sources/mqtt/source.rs

@@ -47,7 +47,9 @@ impl MqttSource {
                 client
                     .subscribe(topic, QoS::AtLeastOnce)
                     .await
-                    .map_err(|_| ())?;
+                    .map_err(|e| {
+                        tracing::error!(error = ?e, "Failed to send MQTT subscribe command");
+                    })?;
             }
             OneOrMany::Many(topics) => {
                 client
@@ -58,7 +60,9 @@ impl MqttSource {
                             .map(|topic| SubscribeFilter::new(topic, QoS::AtLeastOnce)),
                     )
                     .await
-                    .map_err(|_| ())?;
+                    .map_err(|e| {
+                        tracing::error!(error = ?e, "Failed to send MQTT subscribe command");
+                    })?;
             }
         }
 
@@ -80,6 +84,10 @@ impl MqttSource {
                         )) => {
                             // TODO Handle acknowledgement - https://github.com/vectordotdev/vector/issues/21967
                         }
+                        Err(e) => {
+                            tracing::error!("Error = {e:?}");
+                            return Ok(());
+                        }
                         _ => {}
                     }
                 }