Serdar Gökay created CAMEL-26349:
------------------------------------
Summary: camel-paho-mqtt5 - a CamelPahoMqtt5MsgProperties header
shared by several publishes (split, multicast, loop) gets a topic alias written
into it by Paho, so messages arrive on the wrong topic
Key: CAMEL-26349
URL: https://issues.apache.org/jira/browse/CAMEL-26349
Project: Camel
Issue Type: Bug
Components: camel-paho-mqtt5
Affects Versions: 4.23.0
Environment: Camel 4.23.0, Eclipse Paho mqttv5 client 1.2.5, Mosquitto
2.0.22 (default max_topic_alias 10), JDK 25
Reporter: Serdar Gökay
An MQTT-to-MQTT route that consumes from {{paho-mqtt5}}, splits each message
and publishes every part to its own topic with {{CamelPahoMqtt5OverrideTopic}}
delivers some payloads on the wrong topic, with no error and no log line.
The cause is one {{MqttProperties}} object used by several publishes:
# The consumer puts the received message's {{MqttProperties}} object in the
{{CamelPahoMqtt5MsgProperties}} header ({{PahoMqtt5Consumer.createExchange}}).
# Split, multicast, recipient list, loop and wire tap copy header references,
so every publish of the exchange tree sees the same object.
# The producer passes the object from that header to
{{MqttMessage.setProperties}} without copying it
({{PahoMqtt5Producer.process}}).
# Paho 1.2.5 keeps the caller's object ({{MqttAsyncClient.publish}} creates
{{new MqttPublish(topic, message, message.getProperties())}}) and
{{ClientState.send}} writes the outgoing topic alias into it.
# Once the connection has used the Topic Alias Maximum the broker granted in
CONNACK (mosquitto default 10), {{ClientState.send}} assigns no alias for a new
topic but does not clear the alias left by the previous publish. The packet
goes out with the new topic name and that alias, the broker maps the alias to
the new topic, and later alias-only publishes for the old topic are delivered
on the new topic.
A route that publishes once per received message is not affected, because each
received message has its own properties object.
Reproducer (Camel JBang 4.23.0, mosquitto 2.0.22 with its default configuration
on port 18883), repro.camel.yaml:
{noformat}
- route:
id: trigger
from:
uri: timer:trigger
parameters: {period: 1000, repeatCount: 5, delay: 3000}
steps:
- setBody: {constant: "{{items}}"}
- to:
uri: "paho-mqtt5:{{root}}/in/x"
parameters: {brokerUrl: "tcp://127.0.0.1:18883"}
- route:
id: relay
from:
uri: "paho-mqtt5:{{root}}/in/x"
parameters: {brokerUrl: "tcp://127.0.0.1:18883"}
steps:
- convertBodyTo: {type: java.lang.String}
- choice:
when:
- simple: "{{strip}} == true"
steps:
- removeHeader: {name: CamelPahoMqtt5MsgProperties}
- split:
tokenize: ","
steps:
- setHeader:
name: CamelPahoMqtt5OverrideTopic
simple: "{{root}}/out/${body}"
- to:
uri: "paho-mqtt5:{{root}}/out/unrouted"
parameters: {brokerUrl: "tcp://127.0.0.1:18883"}
- route:
id: check
from:
uri: "paho-mqtt5:{{root}}/out/#"
parameters: {brokerUrl: "tcp://127.0.0.1:18883"}
steps:
- convertBodyTo: {type: java.lang.String}
- choice:
when:
- simple: "${header.CamelMqttTopic} == '{{root}}/out/${body}'"
steps:
- log: "RESULT OK topic=${header.CamelMqttTopic} body=${body}"
otherwise:
steps:
- log: "RESULT MISMATCH topic=${header.CamelMqttTopic}
body=${body}"
{noformat}
{noformat}
camel run repro.camel.yaml --max-seconds=20 --prop root=check30 --prop
items=1,2,3,...,30 --prop strip=false
{noformat}
Results (5 batches each; a mismatch is a payload delivered on another item's
topic):
||Items per message||CamelPahoMqtt5MsgProperties||Broker
max_topic_alias||Correct||Wrong topic||
|30|kept|10 (default)|146|4|
|30|removed before split|10 (default)|150|0|
|10|kept|10 (default)|50|0|
|11|kept|10 (default)|51|4|
|30|kept|0|150|0|
With 30 items, from the second batch on the payload for out/10 arrives on
out/30, the last topic that alias 10 was mapped to.
Affected: 4.7.0 (CAMEL-20754) up to 4.23.0 and main.
Suggested fix: the producer builds a new {{MqttProperties}} with only the
publish properties from the header (payload format indicator, message expiry
interval, content type, response topic, correlation data, user properties) and
leaves out the topic alias, which belongs to one connection, and subscription
identifiers, which a client must not send (MQTT 5 section 3.3.4).
{{MqttProperties}} has no copy constructor, so the copy is field by field.
Workaround: remove the {{CamelPahoMqtt5MsgProperties}} header before
publishing, or disable topic aliases on the broker.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
