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,9 @@
+The `mqtt` sink now closes its broker connection after the sink stops, for example when a configuration reload removes or replaces it. Before, the connection stayed open after the sink stopped. A replacement sink that uses the same client ID could then be repeatedly disconnected by the broker.
+
+The sink drains queued messages and QoS exchanges as part of its shutdown. A connection error ends the drain without failing a reload or reconnecting a stopped sink.
+
+Vector's `--graceful-shutdown-limit-secs` or `--no-graceful-shutdown-limit` setting now also controls how long old sinks can drain during a configuration reload. All affected sink tasks share one reload deadline. This applies to all sinks, not only MQTT.
+
+The MQTT source and sink now use `rumqttc-v4-next` 0.34.0 and retain MQTT 3.1.1 support.
+
+authors: thomasqueirozb

src/cli.rs

@@ -237,9 +237,10 @@ pub struct RootOpts {
     #[arg(long, env = "VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT")]
     pub internal_logs_source_rate_limit: Option<NonZeroU64>,
 
-    /// Set the duration in seconds to wait for graceful shutdown after SIGINT or SIGTERM are
-    /// received. After the duration has passed, Vector will force shutdown. To never force
-    /// shutdown, use `--no-graceful-shutdown-limit`.
+    /// Set the duration in seconds to wait for graceful shutdown after SIGINT or SIGTERM.
+    /// Also limit how long old sinks can drain during a configuration reload. After this duration,
+    /// Vector forces shutdown or stops the old sinks. To wait without a limit, use
+    /// `--no-graceful-shutdown-limit`.
     #[arg(
         long,
         default_value = "60",
@@ -248,9 +249,9 @@ pub struct RootOpts {
     )]
     pub graceful_shutdown_limit_secs: NonZeroU64,
 
-    /// Never time out while waiting for graceful shutdown after SIGINT or SIGTERM received.
-    /// This is useful when you would like for Vector to attempt to send data until terminated
-    /// by a SIGKILL. Overrides/cannot be set with `--graceful-shutdown-limit-secs`.
+    /// Never time out while waiting for graceful shutdown or for old sinks to drain during a
+    /// configuration reload. Vector attempts to send data until it is terminated by SIGKILL.
+    /// Overrides/cannot be set with `--graceful-shutdown-limit-secs`.
     #[arg(
         long,
         default_value = "false",

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 {