[jira] [Assigned] (NIFI-16421) ConsumeMQTT (MQTT v3) loses the in-flight message on reconnect with Resume Session: the broker's redelivery arrives before the consuming callback is installed

["Peter Turcsanyi (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/NIFI-16421?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Peter Turcsanyi reassigned NIFI-16421:
--------------------------------------

    Assignee: Peter Turcsanyi

> ConsumeMQTT (MQTT v3) loses the in-flight message on reconnect with Resume Session: the broker's redelivery arrives before the consuming callback is installed
> --------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: NIFI-16421
>                 URL: https://issues.apache.org/jira/browse/NIFI-16421
>             Project: Apache NiFi
>          Issue Type: Bug
>          Components: Extensions
>    Affects Versions: 2.11.0, 2.12.0
>         Environment: MiNiFi Java 2.11.0 in Podman on Ubuntu 24.04. Brokers on the same host: HiveMQ Edge 2025.13 and Eclipse Mosquitto 2. MQTT v3 AUTO (Paho), QoS 1, Session State = Resume Session.
>            Reporter: Mateusz Chrzanowski
>            Assignee: Peter Turcsanyi
>            Priority: Major
>              Labels: mqtt
>
> With {{Session State}} = _Resume Session_ and QoS 1, ConsumeMQTT is expected not to lose messages across a restart: the broker keeps the session and redelivers what was not acknowledged. In practice, *a graceful restart loses exactly one message* whenever a message is in flight at the moment the client disconnects. The message is logged at ERROR level and then acknowledged to the broker, so the broker never redelivers it again:
> {noformat}
> ERROR [MQTT Call: <client id>] o.a.nifi.processors.mqtt.ConsumeMQTT ConsumeMQTT[id=...]
>   MQTT message arrived [topic:test/7/data; payload:[123, 34, 116, ...]]
> {noformat}
> h2. Cause
> In {{PahoMqttClientAdapter}} (nifi-mqtt-processors, {{rel/nifi-2.11.0}}):
> # The constructor calls {{client.setCallback(new DefaultMqttCallback())}}. Its {{messageArrived}} only logs the "MQTT message arrived" line shown above at ERROR and returns.
> # {{ConsumeMQTT.initializeClient()}} calls {{connect()}}, which connects with {{cleanSession = false}}.
> # It then calls {{subscribe()}}, which replaces the callback with {{new ConsumerMqttCallback(handler)}} and subscribes.
> MQTT 3.1.1 section 4.4 requires the server, when a client reconnects with CleanSession = 0, to re-send any unacknowledged PUBLISH packets. The broker does this immediately after CONNACK - between steps 2 and 3. The redelivered message reaches {{DefaultMqttCallback}}, which returns normally, and Paho then acknowledges it (no manual acks are used). The message is logged and discarded; the broker considers it delivered.
> h2. Steps to reproduce
> # An MQTT broker on the same host (reproduced with both HiveMQ Edge 2025.13 and Mosquitto 2).
> # A flow: ConsumeMQTT (MQTT Specification Version {{v3 AUTO}}, QoS {{1}}, Session State {{Resume Session}}, a fixed Client ID, topic filter {{test/+/data}}) -> PublishMQTT republishing every message unchanged to another topic.
> # A publisher sending a continuous stream (100 messages/s here), each message with a unique key, logging what it sent; a subscriber with a persistent session capturing the republished stream.
> # Restart MiNiFi gracefully several times while the stream runs ({{podman restart -t 60}}; the agent stops within about 3 s).
> # Compare what was published with what came out of the flow, key by key.
> h2. Expected
> No message lost across a graceful restart - QoS 1 plus a resumed session.
> h2. Actual - and the same runs with the fix below applied
> Five graceful restarts per run, 100 messages/s; every published message paired with its copy after the flow:
> ||Broker||NiFi MQTT bundle||Published||Lost||Lost per restart||ERROR "MQTT message arrived"||Duplicates||
> |HiveMQ Edge 2025.13|2.11.0 as released|37,254|*5*|1, 1, 1, 1, 1|5 - each one the lost message|0|
> |HiveMQ Edge 2025.13|2.11.0 + fix|37,254|*0*|0, 0, 0, 0, 0|0|0|
> |Mosquitto 2|2.11.0 as released|37,278|*5*|1, 1, 1, 1, 1|5 - each one the lost message|0|
> |Mosquitto 2|2.11.0 + fix|37,260|*0*|0, 0, 0, 0, 0|0|0|
> Every lost message was published within about 1 s after the restart was requested - the one in flight when the client disconnected - and matches the payload of its ERROR line. At a lower rate (48 messages/s, three restarts) two of three restarts lost one message and the third lost none: the loss needs a message in flight at the moment of disconnect, so its frequency depends on the rate.
> Expected to affect a processor stop/start as well (a new client is created and connected before {{subscribe()}}); only agent restarts were tested.
> The same code is in 2.12.0 and on main ({{PahoMqttClientAdapter}} and {{ConsumeMQTT.initializeClient()}} are unchanged as of 2026-10-01).
> h2. Suggested fix
> Install the consuming callback before {{connect()}}. The handler is already known to ConsumeMQTT at that point. A patch - prepared against {{rel/nifi-2.11.0}} and against main; three production files, eleven added lines - adds a default method to the {{MqttClient}} interface, implements it in the Paho adapter, and calls it in {{initializeClient()}} before {{connect()}}:
> {code:java}
> // MqttClient
> default void setReceivedMessageHandler(ReceivedMqttMessageHandler handler) {
> }
> // PahoMqttClientAdapter
> @Override
> public void setReceivedMessageHandler(ReceivedMqttMessageHandler handler) {
>     client.setCallback(new ConsumerMqttCallback(handler));
> }
> // ConsumeMQTT.initializeClient()
> mqttClient = createMqttClient();
> mqttClient.setReceivedMessageHandler(this::handleReceivedMessage);
> mqttClient.connect();
> mqttClient.subscribe(topicPrefix + topicFilter, qos, this::handleReceivedMessage);
> {code}
> The default method is a no-op, so the v5 adapter and every other implementation are unchanged. The internal queue already exists before {{connect()}} (it is created in {{customValidate()}}), so a message delivered before {{subscribe()}} has somewhere to go.
> With it, a new unit test ({{TestConsumeMQTT.testQoS1NotCleanSessionMessageRedeliveredOnConnect}}; the test client redelivers a held message on {{connect()}}) fails without the {{ConsumeMQTT}} change and passes with it; all 41 tests of {{nifi-mqtt-processors}} pass and the {{contrib-check}} profile is clean. A pull request can follow this issue.
> A more complete alternative: {{setManualAcks(true)}} and acknowledge with {{messageArrivedComplete}} only after the message has been accepted. That would also close the loss on a hard kill, but it changes delivery semantics; the fix above does not.
> h2. Side observation
> {{ReceivedMqttMessage.isDuplicate()}} always returns {{false}} (also in 2.12.0: {{ConsumerMqttCallback}} does not pass the DUP flag on), so the {{mqtt.isDuplicate}} attribute and the {{_isDuplicate}} record field never reflect the MQTT DUP flag - a redelivered message cannot be told apart downstream. Separate from this issue; mentioned because it is what one would reach for to verify the fix.
> h2. Related
> * NIFI-10894 - _Resume Session in ConsumeMQTT does not work after NiFi restart_ (1.18/1.19, v5 adapter): the same class of defect - messages delivered before subscribe() are lost. Closed in January 2026 without a fix as part of the 1.x end-of-life cleanup, with a request to re-report for 2.x.
> * NIFI-10885 - ConsumeMQTT v5 client threads prevent a clean shutdown.
> ----
> _This report was drafted with the help of an AI assistant. The measurements, the code analysis and the patch come from my own lab runs and were checked by me._



--
This message was sent by Atlassian Jira
(v8.20.10#820010)