[
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; with manual acks
({{manualAcksEnabled}}) such a message is also acknowledged (QoS 1 in
camel-paho-mqtt5, QoS 1 and 2 in camel-paho; Paho 1.2.5 sends no PUBACK for it
otherwise), so that messages of a stopped consumer with {{cleanStart=false}}
cannot use up the client's {{receiveMaximum}} and stop the broker from sending
to the other routes. 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.
{{messagesOfAStoppedDurableConsumerAreAcknowledgedWithManualAcks}} (shared
client, manual acks, {{receiveMaximum=2}}, route b stopped, three messages for
b, then one for a) fails on the first version of the fix with {{mock://a
Received message count. Expected: <1> but was: <0>}}; in camel-paho the same
test checks that the broker's subscription queue of the stopped consumer
drains. camel-paho-mqtt5 unit tests: 33 run, 0 failures. camel-paho unit tests:
35 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), 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._
> 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; with manual acks
> ({{manualAcksEnabled}}) such a message is also acknowledged (QoS 1 in
> camel-paho-mqtt5, QoS 1 and 2 in camel-paho; Paho 1.2.5 sends no PUBACK for
> it otherwise), so that messages of a stopped consumer with
> {{cleanStart=false}} cannot use up the client's {{receiveMaximum}} and stop
> the broker from sending to the other routes. 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.
> {{messagesOfAStoppedDurableConsumerAreAcknowledgedWithManualAcks}} (shared
> client, manual acks, {{receiveMaximum=2}}, route b stopped, three messages
> for b, then one for a) fails on the first version of the fix with {{mock://a
> Received message count. Expected: <1> but was: <0>}}; in camel-paho the same
> test checks that the broker's subscription queue of the stopped consumer
> drains. camel-paho-mqtt5 unit tests: 33 run, 0 failures. camel-paho unit
> tests: 35 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)
