[thomasqueirozb]

Summary

The MQTT sink's event loop ran in a task detached from the sink. When a reload removed or replaced the sink, that task kept polling and reconnecting, so the old client ID stayed connected to the broker. With rumqttc-next, dropping the last AsyncClient ends the request channel, but the sink merely logged the resulting RequestsDone error and kept polling.

Changes:

• The sink now owns its event loop for its whole lifetime. Topology shutdown waits for the drain, and cancelling or stopping the sink closes its connection.
• The event loop ends on ConnectionError::RequestsDone. The loop is dropped together with the driver's clients, so queued publishes and QoS handshakes complete first.
• An idle sink whose input has ended returns without connecting, using EventLoop::diagnostics to check that its queues are empty.
• After the sink stops, the next connection error ends the drain successfully instead of reconnecting or failing the sink.
• rumqttc-next 0.26.1 is replaced by rumqttc-v4-next 0.34.0. It reports the queue and outbound state used above, so the previous hand-rolled tracking is gone. MQTT 3.1.1 support is kept and TLS stays on Rustls with the Ring provider.
• Retiring sinks in RunningTopology now share one deadline derived from --graceful-shutdown-limit-secs, so a reload no longer waits forever for a sink stuck in a QoS exchange. When the deadline passes, only the sink future is dropped, so unread buffered events remain available to the replacement. --no-graceful-shutdown-limit still waits without a deadline. This applies to every sink, not only MQTT.

References

Related: #26791 (this PR is based on it)

Vector configuration

Manual reload check against a local broker:

data_dir: /tmp
sources:
  input:
    type: stdin
sinks:
  mqtt:
    type: mqtt
    inputs: [input]
    host: 127.0.0.1
    port: 21883
    client_id: reload-check
    topic: shutdown-test
    quality_of_service: exactlyonce
    encoding:
      codec: text

How did you test this PR?

• Unit tests in src/sinks/mqtt/sink.rs against a mock broker: sink_drains_before_returning (QoS 0/1/2), idle_sink_closes_when_input_ends, cancelling_sink_closes_unacknowledged_connection, drain_connection_error_finishes_sink_without_reconnecting, reload_drains_with_vector_shutdown_policy (bounded deadline, unlimited, broker disconnect), drain_handles_queued_publishes_and_reused_ids, drain_after_reconnect (QoS 1/2 resume and reset), and rejected_publish_does_not_block_draining.
• Topology test sink_reload_deadline_preserves_unread_buffers in src/topology/running.rs: two stalled sinks are cancelled at the same deadline, and their unread buffered events reach the replacements.
• Manual check with cargo vdev run against a TCP stub broker and SIGHUP reloads:
  • --graceful-shutdown-limit-secs 2 closed an unacknowledged QoS 2 connection after 2s and started the replacement, which then published on its own connection.
  • Dropping the broker connection mid-drain finished the reload without a panic.
  • --no-graceful-shutdown-limit kept waiting for PUBCOMP and then exited cleanly.
• make test FEATURES="sources-mqtt,sinks-mqtt,sources-prometheus,sinks-prometheus,sources-internal_metrics,sources-splunk_hec,enrichment-tables-memory" SCOPE="-p vector --lib -E 'test(mqtt) or test(topology::)'" passes.
• make check-clippy, make check-markdown, make check-changelog-fragments, and make generate-docs pass.

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-23T21:53:47.023557Z	ad4f710	Draft marked ready

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[datadoghq-integration]

More details

Whether RequestsDone waits for outstanding QoS 2 acknowledgments remains unverified. The available evidence does not establish a delivery regression in that shutdown scenario.

Was this helpful? React 👍 or 👎

_{Open Bits AI session}

_{🤖 Bits Code Review · Commit 34f14fa}

[datadoghq-integration]

Stopping an MQTT sink configured for exactly-once delivery can close its connection before in-flight QoS 2 handshakes finish, leaving messages undelivered before the five-second flush allowance expires.

_{Open Bits AI session}

_{🤖 Bits Code Review · Commit 34f14fa}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 5118bc7ab4

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Bound the drain during configuration reloads

When input ends while a QoS 1/2 publish is awaiting PUBACK/PUBCOMP, this join cannot complete until the broker acknowledges it or the connection errors. For a changed sink whose buffer is reused, src/topology/running.rs:723-726 awaits the old sink without the process-shutdown deadline, so a broker that keeps the connection healthy but withholds the acknowledgement makes the configuration reload hang indefinitely and prevents the replacement sink from starting. Add a bounded/cancellable drain path for reloads rather than relying solely on the process shutdown timeout.

Useful? React with 👍 / 👎.

[thomasqueirozb]

Not bounding this by default is intentional. Reloads that wait for a sink already have no time limit on master. shutdown_diff waits for sinks that reuse their buffer, have a resource conflict, or get a new disk buffer, and it has no timeout. With the default request.retry_attempts, an HTTP-based sink that keeps getting retriable errors already blocks such a reload forever.

With this change, the MQTT sink finishes its in-flight QoS exchanges before it stops. Before, it dropped them. This matches what other sinks do. The drain still ends if the broker closes the connection, if keepalive fails, or if a new connection uses the same client ID.

A default limit would change reload behavior for every sink, so the limit is opt-in through --reload-sink-drain-limit-secs in #26834.

[JinchengCao]

LGTM，you can update the description and test command to reflect the new dependency and the shared reload deadline, which now applies to all sinks

[datadoghq-integration]

More details

MQTT reloads intentionally wait for pending QoS acknowledgments without a deadline. The author confirmed this policy in discussion and tracks an opt-in drain limit separately in #26834.

Was this helpful? React 👍 or 👎

_{Open Bits AI session}

_{🤖 Bits Code Review · Commit ad4f710}