[jira] [Created] (CAMEL-25944) camel-paho-mqtt5 / camel-paho - the consumer sets its callback after connecting, so messages the broker sends for a kept session right after CONNACK are dropped (MQTT 5 QoS 2: lost)

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

shashank created CAMEL-25944:
--------------------------------

             Summary: camel-paho-mqtt5 / camel-paho - the consumer sets its callback after connecting, so messages the broker sends for a kept session right after CONNACK are dropped (MQTT 5 QoS 2: lost)
                 Key: CAMEL-25944
                 URL: https://issues.apache.org/jira/browse/CAMEL-25944
             Project: Camel
          Issue Type: Bug
          Components: camel-paho, camel-paho-mqtt5
            Reporter: shashank
            Assignee: shashank


{{PahoMqtt5Consumer.doStart}} and {{PahoConsumer.doStart}} (main 9f839f4bcc9d) connect the client they created and only then call {{setCallback}}. When the client connects to a session that the broker kept, the broker sends the messages it queued for that session right after CONNACK, before Camel subscribes. A kept session is camel-paho-mqtt5 with {{cleanStart=false}} and a {{sessionExpiryInterval}} greater than 0, or camel-paho with {{cleanSession=false}}, which is the documented way to keep a durable subscription. Paho 1.2.5 handles such a message on its receiver thread while the client may still have no callback. Checked with javap:

* {{CommsCallback.messageArrived(MqttPublish)}} (mqttv5 and mqttv3) queues the message only when an {{mqttCallback}} or a topic listener is set. Otherwise it returns: the message is not delivered and, for QoS 1, not acknowledged (the PUBACK is sent by {{handleMessage}} after delivery);
* mqttv5 {{ClientState.notifyReceivedMsg}} sends PUBREC for a QoS 2 PUBLISH right after {{callback.messageArrived}} whether or not it was delivered, and {{handleInboundPubRel}} sends PUBCOMP. The broker therefore considers the message delivered: *a QoS 2 message (the default {{qos}} of camel-paho-mqtt5) is lost*. A QoS 1 message stays unacknowledged until the client connects again;
* mqttv3 delivers a QoS 2 message on PUBREL and sends PUBCOMP only after delivery, so QoS 1 and QoS 2 messages wait until the next connect. With the default memory persistence they are lost when the client is created again (route or context restart): a PUBREL for an unknown id is answered with PUBCOMP.

The window is small (between CONNACK and {{setCallback}}), but the broker sends the queued messages as soon as the session resumes, so the first messages of a route start after downtime are the ones at risk. Claus Ibsen raised the ordering in the review of #28197 (CAMEL-25793).

*Fix*: set the callback first, then connect the client that the consumer created. A client given with the {{client}} option is connected by the user, as before.

*Tests*: new {{PahoMqtt5ConsumerSessionMessagesTest}} (QoS 2, {{cleanStart=false}}, {{sessionExpiryInterval=3600}}) and {{PahoConsumerSessionMessagesTest}} (QoS 1, {{cleanSession=false}}). In both, a mocked client delivers a message from {{connect()}} to the callback set at that time and drops it when there is none, like Paho 1.2.5 (comment in the test). Both fail on main: {{mock://result Received message count. Expected: <1> but was: <0>}}. camel-paho-mqtt5 unit tests with the fix: 18 run, 0 failures. camel-paho unit tests: 29 run, 8 errors, the same 8 on main: {{PahoComponentTest}} (7) and {{PahoManualAcksTest}} (1) cannot open a socket to the embedded Artemis broker in my sandbox ({{SocketException: Invalid argument}}); CI runs them.

*How it was found*: Claus Ibsen's review comment on #28197, confirmed against the Paho 1.2.5 jars with javap and the JUnit tests above.

*Affected*: main, camel-4.22.x, camel-4.18.x and camel-4.14.x (the same order in {{doStart}}; read, not run).

_Filed with Claude Code on behalf of allthingssecurity._



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)