allthingssecurity opened a new pull request, #28385: URL: https://github.com/apache/camel/pull/28385
# Description [CAMEL-25869](https://issues.apache.org/jira/browse/CAMEL-25869) With `subscriptionType` `Shared`, `Failover` or `Key_Shared`, the consumer creation strategies subscribed `numberOfConsumers` consumers in a loop and caught the `PulsarClientException` of each subscribe, logging it at ERROR. A consumer that could not subscribe was dropped: the route reported Started with fewer consumers than configured, or with none, and nothing subscribed the missing ones again. In the Pulsar client, `subscribe()` throws once the broker could not be reached before the operation timeout (or on an error it does not retry), and the consumer is cleaned up, so there is nothing left that reconnects. With `numberOfConsumers=1` (the default), the route was Started and never consumed a message. The supervising route controller could not help, since the start did not fail. With `Exclusive` the same failure already failed the start of the route. Change: the three strategies share one loop, `CommonCreationStrategyImpl.createMultipleConsumers`. When a subscribe fails, the consumers subscribed so far are closed and the failure is thrown. The route then fails to start, as with `Exclusive`, and can be started again by the caller or the supervising route controller. No option or default changes, so there are no generated changes. What users can notice: a route whose consumer cannot subscribe now fails to start instead of running with missing consumers. When the route starts with the CamelContext, the CamelContext then fails to start, unless the supervising route controller is used, as with `Exclusive` today. A `camel-pulsar` note in the 4.24 upgrade guide says so. With `messageListener=true` (the default) the consumers that did subscribe may already pass messages to the route before a later subscribe fails; they are then closed, so a message they received but did not acknowledge is delivered again after the next successful start (at-least-once); the note says that too. Tests: - New `PulsarConsumerSubscribeFailureTest`, parameterised over `SHARED`, `FAILOVER` and `KEY_SHARED`. A mocked `PulsarClient` throws `PulsarClientException.TimeoutException` on the second subscribe, with `numberOfConsumers=2`. Without the change all three fail with `Expected java.lang.Exception to be thrown, but nothing was thrown`: the route starts with one consumer. With the change the start fails, the consumer that did subscribe is closed, and a second start opens both consumers. - New deterministic simulation test `PulsarConsumerDstTest` (camel-test-dst, package `dst`: `FakePulsar`, `PulsarWorld`), the one that found the bug. The real consumer runs against a fake Pulsar client with subscribe failures (the broker not reachable within the operation timeout), broker restarts, route restarts, suspend/resume and slow or failing processing; an operator starts the route again when its start fails, as the supervising route controller does. It checks that a started route holds the configured number of Pulsar consumers, that every published message is routed, and that the subscription is drained once the faults cleared. 1000 seeds by default; seed 8 (`Key_Shared`, one consumer whose subscribe times out) is pinned with `@DstSeed`. Without the change (the main code of the strategies reverted locally) 871 of 5000 seeds fail, all on "a started route holds the configured number of Pulsar consumers", all with `Shared`, `Failover` or `Key_Shared` and subscribe failures on (215 of them with no consumer at all), and the pinned seed 8 fails; with the change 5000 of 5000 pass. - The test log of camel-pulsar is now at WARN in a size-capped rolling file (50 MB), as a failing world can log a lot. - The worlds table of `test-dst.adoc` (camel-test-dst, and its catalog copy) lists the new world; its intro no longer counts the worlds ("These worlds ..."), the same wording as the kafka-share DST world branch, so the two merge without a conflict. - With the change the camel-pulsar unit tests pass: 1034 run (33, plus 1001 of the DST test), 0 failures, 0 errors, 0 skipped. How it was found: the deterministic simulation above. The property "a started route holds the configured number of Pulsar consumers" failed with `Shared`, `Failover` and `Key_Shared` only, and only when subscribes could fail. In seed 8 (`Key_Shared`, `numberOfConsumers=1`) the route reported Started with no consumer, and none of the 11 published messages was routed in more than 5 simulated minutes after the faults cleared. Without subscribe failures (`-Ddst.pulsar.subscribeFailRates=0`) the code without the change passes, so no other property failed. What was not run: the ITs (they need Docker); a test against a real Pulsar broker; the full root build. # 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-25869](https://issues.apache.org/jira/browse/CAMEL-25869). # 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-pulsar, including the formatter and import-sort plugins. No option metadata 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]
