fix(mqtt sink): stop event loop after the sink stops

Cargo.toml

@@ -481,7 +481,7 @@ rdkafka = { workspace = true, features = ["curl-static", "tokio", "libz", "ssl",
 redis = { version = "0.32.4", default-features = false, features = ["connection-manager", "sentinel", "tokio-comp", "tokio-native-tls-comp"], optional = true }
 regex.workspace = true
 roaring = { version = "0.11.2", default-features = false, features = ["std"], optional = true }
-rumqttc = { package = "rumqttc-next", version = "0.26.1", default-features = false, features = ["use-rustls-no-provider"], optional = true }
+rumqttc = { package = "rumqttc-v4-next", version = "0.34.0", default-features = false, features = ["use-rustls-no-provider"], optional = true }
 # Keep MQTT on Ring alongside Vector's other Rustls users, including in minimal builds.
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"], optional = true }
 rustls-native-certs = { version = "0.8", default-features = false, optional = true }

LICENSE-3rdparty.csv

@@ -66,6 +66,7 @@ async-stream,https://github.com/tokio-rs/async-stream,MIT,Carl Lerche <me@carlle
 async-stream-impl,https://github.com/tokio-rs/async-stream,MIT,Carl Lerche <me@carllerche.com>
 async-task,https://github.com/smol-rs/async-task,Apache-2.0 OR MIT,Stjepan Glavina <stjepang@gmail.com>
 async-trait,https://github.com/dtolnay/async-trait,MIT OR Apache-2.0,David Tolnay <dtolnay@gmail.com>
+async-tungstenite,https://github.com/sdroege/async-tungstenite,MIT,Sebastian Dröge <sebastian@centricular.com>
 asyncband,https://github.com/fast/asyncband,Apache-2.0,The asyncband Authors
 atoi,https://github.com/pacman82/atoi-rs,MIT,Markus Klein
 atomic-waker,https://github.com/smol-rs/atomic-waker,Apache-2.0 OR MIT,"Stjepan Glavina <stjepang@gmail.com>, Contributors to futures-rs"
@@ -504,6 +505,7 @@ mongocrypt,https://github.com/mongodb/libmongocrypt-rust,Apache-2.0,"Abraham Egn
 mongocrypt-sys,https://github.com/mongodb/libmongocrypt-rust,Apache-2.0,"Abraham Egnor <abraham.egnor@mongodb.com>, Isabel Atkinson <isabel.atkinson@mongodb.com>"
 mongodb,https://github.com/mongodb/mongo-rust-driver,Apache-2.0,"Saghm Rossi <saghmrossi@gmail.com>, Patrick Freed <patrick.freed@mongodb.com>, Isabel Atkinson <isabel.atkinson@mongodb.com>, Abraham Egnor <abraham.egnor@mongodb.com>, Kaitlin Mahar <kaitlin.mahar@mongodb.com>, Patrick Meredith <pmeredit@protonmail.com>"
 mongodb-internal-macros,https://github.com/mongodb/mongo-rust-driver,Apache-2.0,"Saghm Rossi <saghmrossi@gmail.com>, Patrick Freed <patrick.freed@mongodb.com>, Isabel Atkinson <isabel.atkinson@mongodb.com>, Abraham Egnor <abraham.egnor@mongodb.com>, Kaitlin Mahar <kaitlin.mahar@mongodb.com>, Patrick Meredith <pmeredit@protonmail.com>"
+mqttbytes-core-next,https://github.com/thehouseisonfire/rumqtt,Apache-2.0,tekjar <raviteja@bytebeam.io>
 multiversion,https://github.com/calebzulawski/multiversion,MIT OR Apache-2.0,Caleb Zulawski <caleb.zulawski@gmail.com>
 multiversion-macros,https://github.com/calebzulawski/multiversion,MIT OR Apache-2.0,Caleb Zulawski <caleb.zulawski@gmail.com>
 multiversion_no_op,https://github.com/hsivonen/multiversion_no_op,Apache-2.0 OR MIT,Henri Sivonen <hsivonen@hsivonen.fi>
@@ -698,7 +700,8 @@ rmpv,https://github.com/3Hren/msgpack-rust,MIT,Evgeny Safronov <division494@gmai
 roaring,https://github.com/RoaringBitmap/roaring-rs,MIT OR Apache-2.0,"Wim Looman <wim@nemo157.com>, Kerollmops <kero@meilisearch.com>"
 roxmltree,https://github.com/RazrFalcon/roxmltree,MIT OR Apache-2.0,Yevhenii Reizner <razrfalcon@gmail.com>
 rsa,https://github.com/RustCrypto/RSA,MIT OR Apache-2.0,"RustCrypto Developers, dignifiedquire <dignifiedquire@gmail.com>"
-rumqttc-next,https://github.com/thehouseisonfire/rumqtt,Apache-2.0,tekjar <raviteja@bytebeam.io>
+rumqttc-core-next,https://github.com/thehouseisonfire/rumqtt,Apache-2.0,tekjar <raviteja@bytebeam.io>
+rumqttc-v4-next,https://github.com/thehouseisonfire/rumqtt,Apache-2.0,tekjar <raviteja@bytebeam.io>
 rust_decimal,https://github.com/paupino/rust-decimal,MIT,Paul Mason <paul@form1.co.nz>
 rustc-hash,https://github.com/rust-lang/rustc-hash,Apache-2.0 OR MIT,The Rust Project Developers
 rustc_version,https://github.com/djc/rustc-version-rs,MIT OR Apache-2.0,The rustc_version Authors

changelog.d/mqtt_sink_reload_connection.fix.md

@@ -0,0 +1,5 @@
+The `mqtt` sink now closes its broker connection when the sink stops, for example when a configuration reload removes or replaces it. Before, the connection stayed open, and the broker could repeatedly disconnect a replacement sink that used the same client ID.
+
+Before it stops, the sink sends its queued messages and completes pending QoS exchanges. A configuration reload that waits for the sink, for example because the replacement reuses its buffer, also waits for this step. If the connection fails during this step, the sink stops without a reconnect and the reload continues.
+
+authors: thomasqueirozb

src/common/mqtt.rs

@@ -116,8 +116,9 @@ impl MqttConnector {
 
     /// Connects the connector and generates a client and eventloop
     pub fn connect(&self) -> (AsyncClient, EventLoop) {
-        let (client, eventloop) = AsyncClient::new(self.options.clone(), 1024);
-        (client, eventloop)
+        AsyncClient::builder(self.options.clone())
+            .capacity(1024)
+            .build()
     }
 
     /// TODO: Right now there is no way to implement the healthcheck properly: <https://github.com/bytebeamio/rumqtt/issues/562>

src/sinks/mqtt/config.rs

@@ -187,13 +187,14 @@ impl MqttSinkConfig {
         }
         let tls =
             MaybeTlsSettings::from_config(self.common.tls.as_ref(), false).context(TlsSnafu)?;
-        let mut options = MqttOptions::new(&client_id, &self.common.host, self.common.port);
+        let mut options =
+            MqttOptions::new(&client_id, (self.common.host.as_str(), self.common.port));
         options.set_keep_alive(self.common.keep_alive);
         options.set_max_packet_size(self.common.max_packet_size, self.common.max_packet_size);
         options.set_clean_session(self.clean_session);
         match (&self.common.user, &self.common.password) {
             (Some(user), Some(password)) => {
-                options.set_credentials(user, password);
+                options.set_credentials(user, password.clone());
             }
             (None, None) => {}
             _ => {

src/sinks/mqtt/integration_tests.rs

@@ -49,12 +49,11 @@ async fn mqtt_happy() {
     // prepare consumer
     let mut mqtt_options = MqttOptions::new(
         "integration-test-consumer",
-        mqtt_broker_address(),
-        mqtt_broker_port(),
+        (mqtt_broker_address(), mqtt_broker_port()),
     );
     mqtt_options.set_keep_alive(5);
 
-    let (client, mut eventloop) = AsyncClient::new(mqtt_options, 10);
+    let (client, mut eventloop) = AsyncClient::builder(mqtt_options).capacity(10).build();
     client
         .subscribe("test", QoS::AtLeastOnce)
         .await

src/sinks/mqtt/service.rs

@@ -2,7 +2,7 @@ use std::task::{Context, Poll};
 
 use bytes::Bytes;
 use futures::future::BoxFuture;
-use rumqttc::{AsyncClient, ClientError};
+use rumqttc::{AsyncClient, ClientError, PublishOptions};
 use snafu::Snafu;
 
 use super::config::MqttQoS;
@@ -80,7 +80,11 @@ impl Service<MqttRequest> for MqttService {
             let byte_size = req.body.len();
 
             let res = client
-                .publish(&req.topic, quality_of_service.into(), retain, req.body)
+                .publish(
+                    req.topic,
+                    req.body,
+                    PublishOptions::new(quality_of_service.into()).retain(retain),
+                )
                 .await;
             match res {
                 Ok(()) => Ok(MqttResponse {