[
https://issues.apache.org/jira/browse/CAMEL-25851?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25851:
-----------------------------
Description:
{{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 that is being stopped
does not subscribe again (as the former camel-paho callback checked with
{{isRunAllowed()}}). 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._
was:
{{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._
> 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
> Priority: Major
>
> {{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
> that is being stopped does not subscribe again (as the former camel-paho
> callback checked with {{isRunAllowed()}}). 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)
