[jira] [Work started] (CAMEL-25851) 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

["shashank (Jira)" <ji...@apache.org>]

     [ 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)