[
https://issues.apache.org/jira/browse/CAMEL-25851?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-25851 started by shashank.
----------------------------------------
> 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 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)
