[
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), also when a single consumer of a shared client is left, as a stopped
consumer with {{cleanStart=false}} leaves its subscription on the connected
client (only a client that the consumer created for itself gives it every
message, as before); a message that no running consumer matches is dropped,
with one WARN per topic and then DEBUG. 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, or another running consumer of the client uses the same
topic filter, as the broker keeps one subscription per filter and client) 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, five 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. Its two newer tests fail on the first version of the fix as well:
{{messagesOfAStoppedDurableConsumerAreNotRoutedToTheConsumerLeft}}
({{cleanStart=false}}, route b stopped) with {{mock://a Body of message: 0.
Expected: <for A> but was: <for stopped B>}}, and
{{stoppingOneOfTwoConsumersOfTheSameTopicKeepsTheOtherSubscribed}} (two routes
of one topic filter, one stopped) with {{mock://a Received message count.
Expected: <1> but was: <0>}}. New {{PahoMqtt5SharedClientReconnectTest}}
(mocked client) fails on main: after {{connectComplete(true, ...)}} only
{{subscribe("topicB", 1)}} is called, {{subscribe("topicA", 1)}} is not. The
same six tests in camel-paho fail the same way. All pass with the fix.
camel-paho-mqtt5 unit tests: 23 run, 0 failures. camel-paho unit tests: 34 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 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._
> 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), also when a single consumer of a shared client is
> left, as a stopped consumer with {{cleanStart=false}} leaves its subscription
> on the connected client (only a client that the consumer created for itself
> gives it every message, as before); a message that no running consumer
> matches is dropped, with one WARN per topic and then DEBUG. 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, or another
> running consumer of the client uses the same topic filter, as the broker
> keeps one subscription per filter and client) 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, five 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. Its two newer tests fail on the first version of the fix as
> well: {{messagesOfAStoppedDurableConsumerAreNotRoutedToTheConsumerLeft}}
> ({{cleanStart=false}}, route b stopped) with {{mock://a Body of message: 0.
> Expected: <for A> but was: <for stopped B>}}, and
> {{stoppingOneOfTwoConsumersOfTheSameTopicKeepsTheOtherSubscribed}} (two
> routes of one topic filter, one stopped) with {{mock://a Received message
> count. Expected: <1> but was: <0>}}. New
> {{PahoMqtt5SharedClientReconnectTest}} (mocked client) fails on main: after
> {{connectComplete(true, ...)}} only {{subscribe("topicB", 1)}} is called,
> {{subscribe("topicA", 1)}} is not. The same six tests in camel-paho fail the
> same way. All pass with the fix. camel-paho-mqtt5 unit tests: 23 run, 0
> failures. camel-paho unit tests: 34 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
(v10.3.26#10030026)
