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)
