shashank created CAMEL-25851:
--------------------------------
Summary: camel-paho-mqtt5 - consumers that share a client (client
option): every message goes to the route that started last, and the others stop
receiving after it stops or the client reconnects
Key: CAMEL-25851
URL: https://issues.apache.org/jira/browse/CAMEL-25851
Project: Camel
Issue Type: Bug
Components: camel-paho, camel-paho-mqtt5
Reporter: shashank
Assignee: shashank
{{PahoMqtt5Consumer.doStart}} (camel-paho-mqtt5, main 20cdb309f7d0, lines
82-133) sets a callback of its own on its {{MqttClient}} with
{{client.setCallback(...)}}. A Paho client has a single callback. With the
{{client}} option ("To use an existing mqtt client"), the consumers of several
endpoints use the same client, for example:
{code:java}
from("paho-mqtt5:sensors/a?client=#mqttClient").to("direct:a");
from("paho-mqtt5:sensors/b?client=#mqttClient").to("direct:b");
{code}
Each consumer subscribes its own topic, but the callback of the consumer that
started last replaces the others, so:
* every message of the client, whatever its topic, is routed by the consumer
that started last: the messages of {{sensors/a}} are processed by the route of
{{sensors/b}};
* when that route stops, its consumer does not unsubscribe (it only does so for
a client of its own) and its callback stays on the client: the other routes
receive nothing more, and the messages of their topics go to the stopped
consumer;
* after the client reconnects, only the consumer that started last subscribes
again ({{connectComplete}}); with a clean session (the default) the broker
dropped the subscriptions of the others, which receive nothing more, while
their routes stay started;
* {{doStop}} sets {{client = null}} also for a client set with the {{client}}
option, so a consumer whose route is started again creates a client of its own
from the endpoint options ({{brokerUrl}} default {{tcp://localhost:1883}})
instead of the shared one.
camel-paho ({{PahoConsumer}}, deprecated) has the same code.
*Fix*: one callback per client, shared by its consumers
({{PahoMqtt5ClientCallback}}, {{PahoClientCallback}} in camel-paho): a consumer
registers on start and unregisters on stop; a message goes to the consumers
whose topic filter matches its topic ({{$share/<group>/}} is ignored for the
match; a client with one consumer gets every message, as before), and every
consumer subscribes again after a reconnect. A consumer of a client set with
the {{client}} option unsubscribes its topic when it stops (unless
{{cleanStart=false}}, {{cleanSession=false}} for camel-paho, as for a client of
its own) and keeps the client for its next start. Upgrade note in the 4.24
guide.
*Tests*: new {{PahoMqtt5SharedClientConsumersTest}} (embedded Artemis broker,
MQTT 5, two routes on one client, three tests) fails on main: {{mock://b Body
of message: 0. Expected: <for B> but was: <for A>}}, {{mock://a Received
message count. Expected: <1> but was: <0>}} after route b stopped, and
{{mock://b Received message count. Expected: <1> but was: <0>}} after route a
was restarted. New {{PahoMqtt5SharedClientReconnectTest}} (mocked client) fails
on main: after {{connectComplete(true, ...)}} only {{subscribe("topicB", 1)}}
is called, {{subscribe("topicA", 1)}} is not. The same four tests in camel-paho
fail on main the same way. All pass with the fix. camel-paho-mqtt5 unit tests:
21 run, 0 failures. camel-paho unit tests: 32 run, 8 errors in
{{PahoComponentTest}} (7) and {{PahoManualAcksTest}} (1), which fail the same
way on main in my environment (the broker address of the test service, a
network interface address, is not reachable there).
*How it was found*: variant analysis of CAMEL-25849 and CAMEL-25850 (a resource
that belongs to one owner kept where another owner finds and replaces or stops
it), then the JUnit tests above.
*Affected*: main, camel-4.18.x and camel-4.14.x (the same {{setCallback}} in
{{doStart}} and {{client = null}} in {{doStop}}; read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
