[
https://issues.apache.org/jira/browse/CAMEL-25851?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25851.
---------------------------------
Resolution: Fixed
> 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
> Fix For: 4.24.0
>
>
> {{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)
