enhancement(mqtt source): support end-to-end acknowledgements#25785
enhancement(mqtt source): support end-to-end acknowledgements#25785antaresindustries wants to merge 8 commits into
Conversation
github-actions
Bot
added
docs review on hold
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 03cb2a37bf
ℹ️ 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".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| entry.connection_generation, | ||
| ) | ||
| { | ||
| pending_acks.try_ack(protocol_state.connected, entry.publish, &client); |
There was a problem hiding this comment.
When acknowledgements are enabled and an earlier publish finalizes as Errored or Rejected, this branch skips that PUBACK but still sends PUBACKs for later Delivered entries from the same connection. That breaks the ordered-ack guarantee this source relies on for QoS 1 publishes: a failed earlier packet remains unacknowledged while later packet IDs are acknowledged, which can make strict brokers disconnect or leave the earlier packet stuck for redelivery while advancing later ones. After the first non-delivered status in a connection generation, later acks for that generation should be withheld until reconnect/redelivery.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed in 8e07a86. Added ProtocolState.ack_suppressed: once a publish in the current connection generation finalizes as anything other than Delivered, every later publish in that generation is also withheld (not just the failed one), resuming normal acking only on the next reconnect. Covered by a new unit test (finalization_suppresses_later_acks_in_generation_after_a_failure).
| if actions.flush_finalizer | ||
| && let Some(finalizer) = &finalizer | ||
| { | ||
| finalizer.flush(); |
There was a problem hiding this comment.
finalizer.flush() only clears receivers that the finalizer stream has already pulled into its ordered set; entries already sent by finalizer.add but still waiting in the finalizer's internal channel survive this flush. After a disconnect/reconnect, a stale pre-disconnect entry can therefore be inserted ahead of new-generation publishes, and because OrderedFinalizer yields in order, a slow or stuck old batch can prevent ACKs for all later redelivered/new MQTT messages even though the generation check would discard that stale entry. Recreate/drop the finalizer or otherwise drain queued entries when the connection generation changes.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed and fixed in 8e07a86. Verified against vector-common's FinalizerSet source: flush() only replaces status_receivers, which the background task hasn't necessarily pulled a concurrent finalizer.add() call into yet -- exactly the race you describe. Replaced both flush() call sites with dropping and recreating the whole finalizer/ack-stream pair (new_finalizer helper): dropping the old finalizer closes its sender, so its background task sees the channel closed and exits, taking any not-yet-picked-up entries with it. No race window.
|
Thanks for the review — both findings were real and are fixed in 8e07a86 (replies inline on each thread with detail):
Both covered by unit tests, and the full suite (unit + live integration against EMQX, 3 repeated runs) is green. |
The mqtt source now participates in Vector's end-to-end acknowledgement framework. When acknowledgements are enabled, the QoS 1 PUBACK is deferred via an OrderedFinalizer until the events are delivered to all sinks; with the source's existing clean_session=false + QoS AtLeastOnce subscription, unacknowledged messages are redelivered after a reconnect (at-least-once). Closes vectordotdev#21967
trying to get ahead of lots of smaller comments on how this works by moving to more state machine like helpers into more dedicated structs
MQTT requires a client to send PUBACK (QoS 1) and PUBREC (QoS 2) packets in the order the corresponding PUBLISH packets were received (MQTT-4.6.0-2). UnorderedFinalizer lets a later batch's ack race ahead of an earlier one still pending delivery, violating this. Switch to OrderedFinalizer, same as the kafka source's ordered offset commits: a slow/stuck earlier batch now holds back acks for publishes received after it, trading a little head-of-line latency for protocol compliance.
…izer on reconnect Addressed PR review findings: - Once a publish in the current connection generation finalizes as anything other than Delivered, withhold acks for every later publish in that same generation too, not just the failed one. MQTT requires PUBACKs to be sent in the order their publishes were received (MQTT-4.6.0-2); acking a later packet while an earlier one was never acked would violate that order. The withheld ones are redelivered the same way the one that failed already is. - Replace finalizer.flush() with recreating the finalizer/ack-stream pair on (re)connect. flush() only clears entries already pulled into its internal ordered set; an entry added concurrently (via finalizer.add) that the finalizer's background task hadn't yet picked up survives the flush and gets pushed into the fresh set anyway. Since OrderedFinalizer won't yield anything newer until that stale entry resolves, it could hold back acks for the new generation. Dropping the old finalizer outright removes the race: its background task sees the channel closed and exits, taking any not-yet-picked-up entries with it.
antaresindustries
force-pushed
the
mqtt-source-e2e-acknowledgements
branch
from
8e07a86 to
986ce0c
Compare
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 986ce0c24a
ℹ️ 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".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Err(_) => { | ||
| let actions = protocol_state.on_disconnect(); | ||
| if actions.clear_pending_acks { | ||
| pending_acks.clear(); | ||
| } | ||
| if actions.recreate_finalizer { | ||
| (finalizer, ack_stream) = self.new_finalizer(&shutdown); |
There was a problem hiding this comment.
When acknowledgements are enabled and the socket drops after client.try_ack has queued a PUBACK/PUBREC but before the event loop writes it, connection.poll() cleans the event loop by moving unread requests into its pending queue for replay on the next connection. This branch only clears Vector's pending_acks, so that stale ack can be sent immediately after the next ConnAck; if the broker starts a fresh session or reuses the packet id, it can acknowledge the wrong publish or trigger a protocol disconnect. Filter ack requests out of the event loop's pending queue here as well.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed via the vendored rumqttc source: EventLoop::poll() calls self.clean() internally before returning this error, and clean() moves any not-yet-flushed Request::PubAck/Request::PubRec into the public connection.pending: VecDeque<Request> field for automatic replay on the next connection. Fixed by filtering those out of connection.pending alongside clearing Vector's own pending_acks, in the same disconnect-handling branch. Pushed in 4dcc85b.
| if actions.resubscribe { | ||
| protocol_state.on_resubscribe_result(self.try_subscribe(&client)); |
There was a problem hiding this comment.
Here a fresh-session reconnect is considered resubscribed as soon as try_subscribe queues the request, but the broker has not accepted it until a matching SUBACK is received. If the connection drops after this queued SUBSCRIBE is lost but before the broker processes it, the next reconnect can report session_present = true for an empty persistent session, so actions.resubscribe is false and pending_resubscribe has already been cleared; the source then stays connected without any subscriptions and stops receiving messages. Keep the resubscribe pending until SUBACK confirms success, or resend after reconnects until confirmed.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Right — queueing locally only means rumqttc accepted the request, not that the broker processed it, and that gap is exactly the race you describe. Fixed: pending_resubscribe now only clears on an actual SUBACK (a new resubscribe_awaiting_suback flag stops the loop from resending every iteration while one is outstanding, without yet treating it as done), and if a reconnect's session_present comes back true while a previous attempt was never confirmed, it stays pending regardless — added a test (resubscribe_stays_pending_across_a_session_present_reconnect_without_suback) for that exact scenario. Pushed in 4dcc85b.
| if status != BatchStatus::Delivered { | ||
| self.ack_suppressed = true; | ||
| return false; |
There was a problem hiding this comment.
The new ack_suppressed path fixes the earlier ordering issue, but for brokers that keep QoS 1 messages in flight until the session reconnects, this branch leaves the connection open after an Errored or Rejected batch. The failed publish is never PUBACKed, later publishes are also suppressed, and the broker can simply fill its in-flight window and stop delivering rather than redelivering the failed packet. Force a reconnect (or another explicit redelivery path) when a current-generation publish is withheld.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Agreed this was a real gap — chose to fix it now rather than defer, but carefully, given the earlier lesson on this same feature that forcing a reconnect via an abrupt EventLoop::clean() broke against a live EMQX broker (unexpected_sock_close, session stopped delivering entirely — see the closed PR #25770's history).
This time: the first time an ack is withheld in a generation, send a graceful MQTT Disconnect via client.try_disconnect() instead of dropping the network directly — a client-initiated Disconnect is a normal, expected close from the broker's perspective, not the abrupt-socket-close condition that tripped EMQX before. Rate-limited to at most once per 30s (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure (the redelivered publish failing again immediately) can't thrash the connection.
Verified against a live EMQX broker with a new integration test (mqtt_forces_reconnect_after_withheld_ack: sink rejects the first delivery attempt only, so the only way a second delivery of the same message shows up is a broker redelivery following a reconnect the source forced itself) — reran 4x stable, ~1.4s/run, no unexpected_sock_close this time. Pushed in 4dcc85b.
…econnect on withheld ack Three fixes to reconnect/ack handling, all found by review on vectordotdev#25785: - On disconnect, rumqttc's own `EventLoop::clean()` moves any PUBACK/PUBREC that `try_ack` queued but hadn't reached the network yet into `connection.pending` for automatic replay on the next connection. Those reference packet ids from the connection that just died, so filter them out alongside clearing Vector's own `pending_acks`, or the broker could receive a stale ack for the wrong publish once reconnected. - A (re)subscribe was considered done as soon as `try_subscribe` queued the request locally, not once the broker actually confirmed it via SUBACK. If that queued request is lost (a disconnect racing the send), a later reconnect can report `session_present = true` for a session that never got the subscription, leaving the source connected but receiving nothing. `pending_resubscribe` now only clears on an actual SUBACK, and survives a `session_present = true` reconnect if the previous attempt was never confirmed. - Withholding a PUBACK for ordering correctness left the connection open indefinitely. Brokers that only redeliver in-flight QoS 1 messages on reconnect (not on an independent timer) would then just fill their in-flight window and stop delivering once withheld, with no way to recover. Force a reconnect the first time an ack is withheld in a generation, rate-limited (FORCED_RECONNECT_COOLDOWN) so a persistent downstream failure can't thrash the connection. Uses a graceful `try_disconnect()` rather than dropping the network directly: verified against a live EMQX broker that the latter triggers a broker-side `unexpected_sock_close` error that stops delivery entirely, whereas an explicit `Disconnect` is treated as a normal close. Verified against a live EMQX broker (including a new mqtt_forces_reconnect_after_withheld_ack integration test), rerun 4x stable.
|
Fixed all three new review findings in 4dcc85b:
All existing + new unit tests, clippy, and the full live-broker integration suite pass. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4dcc85b8f2
ℹ️ 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".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // normal, expected client-initiated close. | ||
| if decision.just_suppressed { | ||
| let now = tokio::time::Instant::now(); | ||
| if forced_reconnect_cooldown_until.is_none_or(|until| now >= until) { |
There was a problem hiding this comment.
When a redelivered publish is rejected again before FORCED_RECONNECT_COOLDOWN expires, this condition skips queuing a disconnect and nothing schedules one for when the cooldown ends. At that point ack_suppressed remains set for the current generation, the broker can keep the unacked QoS 1 publish in flight, and with no later finalizer event the source stays connected indefinitely instead of triggering the reconnect needed for redelivery. The fresh issue is the cooldown branch itself: after it evaluates false, there is no delayed reconnect path.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed — after the cooldown branch evaluated false there was no delayed reconnect path, and since just_suppressed fires only once per generation, nothing would ever re-trigger it: the source stayed connected and stuck indefinitely. Fixed in 03d857e by making the reconnect scheduled rather than conditional: a suppression inside the cooldown window now sets the reconnect's due time to cooldown expiry, and a dedicated sleep_until branch in the select loop fires it (new ForcedReconnectSchedule, unit-tested for exactly this scenario in forced_reconnect_inside_cooldown_is_deferred_not_skipped).
Since this was the second finding of the same shape (a needed action with no wake-up path), the same commit also fixes six more issues found by a deeper audit along that pattern and its siblings — details in the summary comment below.
… holes and cross-layer staleness One fix for the latest review finding, plus six more from a deeper audit of the same failure classes (states whose required action has no wake-up path, Vector state diverging from rumqttc's internals, and locally-queued mistaken for peer-confirmed): - A withheld-ack forced reconnect landing inside the cooldown window is now scheduled for cooldown expiry instead of skipped: `just_suppressed` fires only once per generation, so a skipped reconnect never got another trigger and the source stayed connected (and stuck) indefinitely. - On disconnect, clear rumqttc's buffered `state.events`: `clean()` doesn't, so packets batched from the dead connection were yielded after the next ConnAck — a stale Publish got tagged with the new generation (its stale packet id then acked on the new connection), and a stale SubAck falsely confirmed the new connection's subscribe. Also drop a queued `Request::Disconnect` from `connection.pending`, which would otherwise be replayed onto (and immediately kill) the next connection. - SUBACK failure return codes (0x80, e.g. an ACL denial) no longer count as subscribed: the rejected topic is logged at error level (return codes map to topics by order, [MQTT-3.9.3-1]) and the (re)subscribe is retried on a 30s pace instead of being silently abandoned. - The initial SUBSCRIBE now goes through the same SUBACK-confirmation tracking as re-subscribes (issued on the first ConnAck, driven by `pending_resubscribe` starting true). Previously a lost initial SUBSCRIBE followed by a session_present reconnect left the source permanently subscribed to nothing. - The forced reconnect no longer assumes the broker closes the socket after our DISCONNECT (rumqttc never closes its side; the server only SHOULD): if no disconnect is observed within 10s, drop the connection locally, which [MQTT-3.14.4-2] requires of the client anyway. While a DISCONNECT is unanswered, nothing further is written to the connection ([MQTT-3.14.4-1]). - A newly finalized PUBACK now queues behind earlier acks still parked for retry instead of potentially jumping ahead of them ([MQTT-4.6.0-2]). - `keep_alive = 0` is rejected when acknowledgements are enabled: with the heartbeat disabled, a silently dead connection is never detected on a quiet topic, so unacked messages would never be redelivered. Verified against a live EMQX broker: all 4 integration tests (including the forced-reconnect redelivery test) rerun 4x stable; 20 unit tests cover the new state transitions.
|
03d857e fixes the latest review finding (a forced reconnect suppressed inside the cooldown window was skipped with no rescheduling path — now scheduled for cooldown expiry via a timer branch) plus six more issues found by a deeper audit targeting the same failure classes. All were verified against the rumqttc 0.24 source and, where applicable, a live EMQX broker:
Verified: 20 unit tests (including new coverage for each state-machine change), clippy clean, and the full live-broker integration suite (4 tests including forced-reconnect redelivery) rerun 4x stable against EMQX 5.0.15. Known limitations documented rather than changed: prolonged sink backpressure still blocks the poll loop past the keepalive deadline (self-healing via redelivery — inherent to the single-task backpressure design this PR deliberately keeps), and rumqttc's |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 03d857e8d8
ℹ️ 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".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // dead connection -- the reconnect ConnAck is returned directly | ||
| // by `poll()`, never through this buffer -- so drop it all. Any | ||
| // dropped publish is unacked and will be redelivered. | ||
| connection.state.events.clear(); |
There was a problem hiding this comment.
When acknowledgements are off (the default), rumqttc is still in auto-ack mode, so publishes sitting in state.events can already have had their PUBACK queued or sent before this error path runs. Clearing the buffer here drops those messages instead of delivering them, and a broker that has observed the auto-ack is allowed not to redeliver them; only discard these stale buffered publishes when manual acknowledgements are enabled, or otherwise drain them for the default path.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Confirmed — in auto-ack mode handle_incoming_publish queues the PUBACK the moment it buffers the publish into state.events, so it can already be on the wire when the error path runs, and clearing the buffer would drop a message the broker is allowed to never redeliver. Fixed in f06541d: the scrub is now mode-dependent. Manual-ack mode (acknowledgements on) still clears everything — nothing buffered has been acked there, so the broker redelivers it all, and dropping prevents the stale-generation ack and stale-SubAck problems. Auto-ack mode retains buffered publishes for processing and drops only the non-publish events, so a stale SubAck still can't falsely confirm the new connection's subscribe.
| OneOrMany::One(topic) => client | ||
| .try_subscribe(topic, SUBSCRIPTION_QOS) | ||
| .map_err(|_| ()), |
There was a problem hiding this comment.
When a configured topic filter is invalid (for example foo/#/bar), rumqttc rejects try_subscribe before enqueueing the request, but this path collapses that permanent configuration error into the same Err(()) used for a full request channel. Because pending_resubscribe remains set, the source now retries every loop while connected instead of failing the source as the previous awaited subscribe path did, leaving users with a running component that receives nothing and log-spams until the config is fixed.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
The user impact here is real and now fixed, with one correction on the mechanism: in rumqttc 0.24, try_subscribe does not validate the filter at all — it's a bare try_send (client.rs:160-165), and outgoing_subscribe only rejects empty filter lists — so an invalid filter was actually reaching the broker, surfacing as a SUBACK failure (now caught by the paced-retry path, but still a permanently broken source) or a protocol disconnect, not as a local queueing error. The previous awaited subscribe path had the same gap: subscribe().await also only fails on channel errors, so it never failed fast on a bad filter either.
Fixed in f06541d one level earlier than either path: topic filters are validated with rumqttc::valid_filter when the source is built, so an invalid filter (e.g. foo/#/bar) fails config load with an error naming the bad filter, before anything connects. The runtime retry path is untouched — with filters validated at build, a try_subscribe error there can only mean a full request channel, which is genuinely transient and correct to retry.
…eject invalid topic filters at build Two fixes from review on vectordotdev#25785: - The disconnect-time scrub of rumqttc's buffered `state.events` was unconditional, but in auto-ack mode (acknowledgements off, the default) rumqttc queues the PUBACK the moment it buffers a publish -- it may already be on the wire, and the broker is then allowed to never redeliver. Dropping such a publish would lose it outright. The scrub is now mode-dependent: manual-ack mode still drops everything (nothing buffered has been acked, so the broker redelivers it all), while auto-ack mode keeps buffered publishes for processing and drops only the non-publish events (so e.g. a stale SubAck still can't falsely confirm the new connection's subscribe). - An invalid topic filter (e.g. `foo/#/bar`) is a permanent configuration error, but rumqttc queues the SUBSCRIBE without validating it, so it only surfaced at runtime as a source that receives nothing while the broker rejects the subscription over and over. Topic filters are now validated with `rumqttc::valid_filter` when the source is built, failing fast with an error naming the bad filter.
…e, fault-inject against a live broker Review on vectordotdev#25785 kept finding bugs in the gaps between hand-picked test cases: event sequences the unit tests didn't sample, and broker-side failure paths the integration tests never exercised (every existing test used acks-on, client-side aborts only). Close both gaps structurally instead of sampling harder: - `exhaustive_protocol_state_model_check` BFS-explores every reachable abstract state of `ProtocolState` + `ForcedReconnectSchedule` + the resubscribe retry timer under every legal event interleaving (timer state abstracted to armed/not-armed, so the space is finite), asserting on every transition the invariants past findings violated: a suppressed generation always has a recovery action armed, per-connection schedules die with the connection, an unconfirmed subscription always has a live retry path, `pending_resubscribe` is only cleared by a fully granted SUBACK, no write after an unanswered Disconnect. Mutation-verified: reintroducing either the cooldown-skip bug or the session_present resubscribe-clearing bug makes it fail with a minimal counterexample trace. - Three fault-injection integration tests force *server-side* disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2], portable across brokers, no admin API): recovery with the session preserved (offline publishes delivered from the session queue), resubscribing after the broker discards the session entirely (`session_present = false` path, SUBACK-confirmed), and no message loss across a kick in the default auto-ack mode -- the mode and the disconnect direction that previously had zero coverage. All 29 tests pass against a live EMQX broker, rerun 5x stable.
|
3e45f16 adds two structural test layers aimed at ending the find-fix-review cycle, rather than waiting for the next round of findings: Exhaustive state-machine model check ( Fault-injection integration tests: all prior integration coverage used client-side aborts with acknowledgements on. Three new tests force server-side disconnects via MQTT client-ID takeover ([MQTT-3.1.4-2] — portable, no broker admin API): recovery with the session preserved, resubscribing after the broker discards the session entirely (the All 29 tests pass against live EMQX 5.0.15, rerun 5x stable. |
1 participant
Summary
Makes the
mqttsource participate in Vector's end-to-end acknowledgementframework, so it can provide at-least-once delivery instead of best-effort.
acknowledgementsoption (SourceAcknowledgementsConfig, the standardbool_or_structform used by other sources) and flipscan_acknowledge()totrue.MqttOptions::set_manual_acks(true)so rumqttc stops auto-acking incomingpublishes.
BatchNotifier; the publish is registered with anOrderedFinalizer, and the QoS 1PUBACKis sent (client.ack(&publish))only once the batch reaches
BatchStatus::Delivered. OnErrored/Rejectedthe message is left unacked. Finalization is ordered (not just per-packet
independent) because MQTT requires PUBACKs to be sent in the order their
publishes were received (MQTT-4.6.0-2), the same reason the
kafkasourceuses ordered offset commits.
would otherwise risk deadlocking itself if it awaited
client.ack/client.subscribedirectly (the same task also has to drain rumqttc'soutgoing request channel that those calls feed into), it uses the
non-blocking
try_ack/try_subscribevariants plus a small retry queue(
PendingAcks) for anything that couldn't be sent immediately, and retriesresubscribing after a reconnect until it succeeds. This keeps polling and
processing coupled: when a downstream sink can't keep up, this task simply
stops polling, which throttles the broker via ordinary TCP backpressure —
no message is ever buffered or dropped locally, and no broker-specific
behavior is assumed anywhere.
Why
The source previously created a
BatchNotifierbut discarded the receiver andauto-acked every message on arrival, so events could be lost if Vector or a
downstream sink failed before the write completed. Because the source already
uses
clean_session = falseand subscribes at QoSAtLeastOnce, deferring thePUBACKlets the broker redeliver any unacknowledged message after a reconnect —giving genuine at-least-once delivery. This is the work tracked in #21967.
A note on the design
An earlier attempt at this split polling and processing across two tasks, so
the main task could use blocking
client.ack/client.subscribecalls withoutneeding the retry-queue machinery described above. That's a real simplification
in the ack/subscribe path, but it requires bounding a channel between the two
tasks that can't safely block without risking a deadlock — so it either drops
messages under sustained backpressure (relying on the broker to redelivered
them, which the MQTT spec only guarantees on reconnect, not on an open
connection) or grows memory unboundedly. Verified against a live EMQX broker
that forcing a reconnect to close that gap introduces a different, worse
failure mode (see the history on the linked PR for detail). Given the whole
point of
acknowledgementsis a delivery guarantee, this PR keeps thesingle-task design: more code in the ack/subscribe path, but it never drops or
buffers a message locally under any downstream condition.
Behavior / compatibility
acknowledgementsoff → manual acks off → rumqttcauto-acks exactly as before). Opt-in only.
This matches the guarantee of other ack-capable sources.
Vector configuration
How did you test this PR?
cargo check/cargo clippy -D warnings/cargo fmt --checkonsrc/sources/mqtt/*.rs— clean.cargo test -p vector --lib --no-default-features --features sources-mqtt sources::mqtt— all pass, including thePendingAcksretry/backpressure matrix and protocol-state contract tests.
cargo vdev int start mqtt/cargo vdev int test mqtt): all 3 tests pass, includingmqtt_redelivers_unacknowledged_messages, which forces a sink rejection toprove a message is redelivered on reconnect. Reran the full suite 3 times —
stable, ~1s/run.
(
website/cue/reference/components/sources/generated/mqtt.cuegains thestandard
acknowledgementsblock).(
changelog.d/21967_mqtt_source_acknowledgements.enhancement.md).Change Type
Is this a breaking change?
Does this PR include user facing changes?
changelog.d/21967_mqtt_source_acknowledgements.enhancement.mdReferences
Incoming::PubAckin themqttsource #21967the reasons above)