allthingssecurity opened a new pull request, #28324: URL: https://github.com/apache/camel/pull/28324
# Description [CAMEL-25851](https://issues.apache.org/jira/browse/CAMEL-25851) A Paho `MqttClient` has a single callback, and `PahoMqtt5Consumer` set one of its own on its client. With the `client` option, the consumers of several endpoints use the same client, so the consumer that started last received every message of the client: the messages of topic A were routed by the route of topic B. When that route stopped, the other routes received nothing more (its callback stayed on the client), and after the client reconnected only that consumer subscribed again, so with a clean session the topics of the other consumers were lost for good while their routes stayed started. A stopped consumer also dropped the shared client, so it created a client of its own (from the endpoint options) when its route started again. `PahoConsumer` in camel-paho has the same code. Changes (camel-paho-mqtt5, and the same in camel-paho): - New `PahoMqtt5ClientCallback` (`PahoClientCallback`): one callback per client, shared by the consumers of that client. A message goes to the consumers whose topic filter matches its topic (`$share/<group>/` is left out of the match), also when a single consumer of a shared client is left: a stopped consumer with `cleanStart=false` leaves its subscription on the connected client, so its messages must not go to the consumer left. 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 `manualAcksEnabled=true` a dropped message of a shared client is also acknowledged (`messageArrivedComplete`): Paho sends no PUBACK for it otherwise (checked with javap in Paho 1.2.5: `CommsCallback.handleMessage` sends the PUBACK, and in camel-paho the PUBCOMP, only when manual acks are off), and after `receiveMaximum` such messages (camel-paho: the broker's limit, e.g. m osquitto `max_inflight_messages`) the broker sent nothing more to any consumer of the client, the failure mode of CAMEL-25793. In camel-paho-mqtt5 only a QoS 1 message is acknowledged there, as Paho completes the QoS 2 flow on its own; the consumer's own client is not, as it is being stopped and a kept session sends the message again on the next connect. The callback iterates a snapshot of its consumers. On a reconnect every consumer subscribes again (`resubscribe`, the former `connectComplete` code, including the route restart for a client of its own); a consumer that is being stopped does not (`isRunAllowed()`, as the former camel-paho callback checked). In camel-paho-mqtt5 the callback hands every `connectComplete` to each consumer, so the reconnect a consumer makes itself after a failed message with `manualAcksEnabled` (CAMEL-25793) still subscribes again only when the broker did not keep the session, as on main. - A consumer registers on start and unregisters on stop and on a failed start. A consumer of a shared client unregisters before it unsubscribes its topic; a consumer's own client is unregistered only after its disconnect, so that its messages are routed until then, as before. - A consumer of a client set with the `client` option unsubscribes its topic when it stops (unless `cleanStart=false`, `cleanSession=false` for camel-paho, the same rule as for a client of its own, or another running consumer of the client uses the same topic filter: the broker keeps one subscription per filter and client, so unsubscribing it would stop the other route), and keeps the client, so that it uses it again when its route starts again. The rule reads the endpoint option, not how the external client was connected; the upgrade note says so. - Limits, now in the upgrade guide entry: with `cleanStart=false` (`cleanSession=false`) a stopped consumer keeps its subscription on the connected client, so the broker keeps sending its messages, which are dropped until it starts again (on main they went to the stopped consumer's callback). Consumers of one client whose topic filters overlap (`orders/#` and `orders/eu`) each get a message that matches several filters, and a broker that sends one copy per matching subscription makes each of them get every copy: checked against the embedded Artemis of the tests, one message published to `ov/x` with consumers of `ov/#` and `ov/x` on one client reached each route twice. MQTT 5 subscription identifiers (Paho assigns one per `subscribe` with a message listener) could tell the copies apart; that is a possible follow-up, not part of this PR. With `manualAcksEnabled=true` each consumer that gets a message acknowledges it (Paho sends a PUBACK per `messageArrivedComplete` call), so a messa ge that failed in one of these routes is still acknowledged when another of them processes it. Each `subscribe` of a filter makes the broker send the retained message of a matching topic again (Retain Handling 0), so consumers of the same filter get it again when another of them starts and after each resubscribe on a reconnect. Both are in the upgrade note; neither is changed here. - Upgrade guide entry (4.24) for the unsubscribe on stop and the reuse of the client. - No pom change: the test dependencies of the embedded broker (`artemis-mqtt-protocol`, `camel-test-infra-artemis`) are on main since CAMEL-25793. Tests: - New `PahoMqtt5SharedClientConsumersTest` (embedded Artemis, MQTT 5, two routes on one client). Without the change: `eachConsumerReceivesItsOwnTopic` fails with `mock://b Body of message: 0. Expected: <for B> but was: <for A>`, `stoppedConsumerNoLongerReceivesAndOtherConsumerKeepsItsTopic` with `mock://a Received message count. Expected: <1> but was: <0>`, `restartedConsumerKeepsUsingTheSharedClient` with `mock://b Received message count. Expected: <1> but was: <0>`. Two more tests fail without the change and also on the first version of this PR (found in review): `messagesOfAStoppedDurableConsumerAreNotRoutedToTheConsumerLeft` (`cleanStart=false`, route b stopped, then one message for b and one for a) with `mock://a Body of message: 0. Expected: <for A> but was: <for stopped B>`, and `stoppingOneOfTwoConsumersOfTheSameTopicKeepsTheOtherSubscribed` (routes `sharedA` and `sharedA&qos=1` on one client, route b stopped) with `mock://a Received message count. Expected: <1> but was: < 0>`. One more test fails without the change and on the previous version of this PR (found in review): `messagesOfAStoppedDurableConsumerAreAcknowledgedWithManualAcks` (a shared client with `setManualAcks(true)` and `receiveMaximum=2`, endpoints with `qos=1&manualAcksEnabled=true&cleanStart=false`, route b stopped, three messages for b and then one for a) with `mock://a Received message count. Expected: <1> but was: <0>`: the broker sends nothing more once two dropped messages are unacknowledged. - New `PahoMqtt5SharedClientReconnectTest` (mocked client): without the change only `subscribe("topicB", 1)` is called after `connectComplete(true, ...)`. - The same seven tests in camel-paho (`PahoSharedClientConsumersTest`, `PahoSharedClientReconnectTest`) fail the same way without the change, except the manual-ack test: MQTT 3.1.1 has no receive maximum and the embedded Artemis sets no in-flight limit for it, so the test checks that the dropped message is gone from the broker's queue of the subscription (an unacknowledged message stays there), and fails without the change with `expected: 0L but was: 1L within 10 seconds`. All pass with the change. - camel-paho-mqtt5 unit tests: 33 run, 0 failures (including the 9 `PahoMqtt5ManualAcksFailureTest` tests of CAMEL-25793). camel-paho unit tests: 35 run, 8 errors in `PahoComponentTest` (7) and `PahoManualAcksTest` (1), which fail the same way without the change in my environment (28 run, the same 8 errors): they connect to the address of a network interface given by the test service, which is not reachable there. The new tests use `localhost`. 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. What was not run: the full root build, and a reconnect against a real broker (covered by the mocked client test). # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-25851](https://issues.apache.org/jira/browse/CAMEL-25851). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested camel-paho and camel-paho-mqtt5. No option changed. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
