allthingssecurity opened a new pull request, #28810: URL: https://github.com/apache/camel/pull/28810
# Description [CAMEL-26453](https://issues.apache.org/jira/browse/CAMEL-26453) `DynamicRouterEndpoint.doInit()` creates the `DynamicRouterProcessor` of its channel and adds it to the component with `addRoutingProcessor(channel, processor)`. That fails when the channel already has a processor (`Dynamic Router can have only one processor per channel; channel 'orders' already has a processor`). Nothing removed the processor again, except the component's `doShutdown`. When a route that sends to `dynamic-router:orders` is removed, Camel stops and removes the endpoints that only that route used. The endpoint was gone, but its processor stayed in the component. Then: - Adding the route again failed with `FailedToCreateRouteException ... already has a processor`. A route reload in development mode replaces the routes of a changed file by removing them and adding them again, and fails the same way (a probe that reloads the route through `updateRoutesToCamelContext` fails on main and works with this change). - A `toD("dynamic-router:${header.channel}")` in another route had used the same endpoint while the removed route existed (a `toD` does not keep an endpoint in use). On its next message the `toD` created the endpoint again, and every exchange failed with `ResolveEndpointFailedException`. - The same happened when the producer cache of a `toD` evicted the endpoint and a later message created it again. The subscriptions of the channel are kept by the component (the filter service). So the channel works again as soon as the new endpoint can use a processor for it. The same check also made a stopped endpoint fail when it was started again. `BaseService.start` runs `init` again after a stop, so `doInit` tried to add a second processor for its channel. Changes: - The component keeps one processor per channel, together with the endpoint that uses it. `DynamicRouterEndpoint.doInit` asks the component for the processor of its channel (`getOrCreateRoutingProcessor`): - An endpoint with the same URI as the endpoint that uses the processor takes the same processor. This covers a route that is removed and added again, a route reload, a `toD` that creates the endpoint again after it was removed or evicted, and an endpoint that is started again. - An endpoint with another URI for the channel still fails while the endpoint that uses the processor is in use, as before. - Once that endpoint has been stopped (with its route), the processor is replaced by one for the new endpoint. An example is a route added again with other options. - The new processor is created first, so the channel keeps its processor when that fails. - The old processor's `RecipientList` is then stopped. Its `parallelProcessing` pool is only shut down with the component, so exchanges that another route still has in flight in it (for example a `toD` that used the old URI) are not interrupted or dropped. Stopping the recipient list does stop its producer cache, so such an exchange can fail with an exception if its destination producer needs a connection; with `direct:` destinations it completes. - Nothing is stopped when an endpoint stops; it only marks the processor as replaceable. So a send that still uses an endpoint that was removed or evicted keeps routing. No processor (or `RecipientList` with its `parallelProcessing` pool) is created per re-creation of an endpoint with the same URI. - When the channel has no processor, the producer fails the exchange with an `IllegalStateException` that names the channel, instead of a `NullPointerException`. - `DynamicRouterComponent.doShutdown` passed the map itself to `ServiceHelper.stopAndShutdownService`, which stopped nothing. It now stops and shuts down the recipient lists of the processors, the replaced ones included. - `addRoutingProcessor` (package-private) is removed, as nothing calls it any more. Its two unit tests go with it; the get-or-create test covers both cases. - The processors map is a `ConcurrentHashMap`, and the get-or-create runs under a lock. Tests: - `DynamicRouterRemoveRouteTest` (new, 9 tests): - remove the route and add it again: messages before and after reach the subscription, with the same processor; - add it again with other options: a new processor, and the old recipient list is stopped; - the same with `parallelProcessing=true`: the thread pool of the replaced processor is not shut down; - a `toD` after the route that used the endpoint was removed; - a stop and start of the endpoint keeps the processor; - a second endpoint for the channel, while the first is in use, fails and leaves the processor in place (not stopped); - a send through an endpoint that was removed still routes, asynchronous and `synchronous=true`; - with `parallelProcessing=true`, an exchange in flight when the endpoint is removed completes. - `DynamicRouterComponentTest`: the get-or-create rules with mock endpoints: - the same URI gets the same processor; - another URI fails while the first endpoint is in use; - a failing factory keeps the old processor (not stopped); - after a stop, a replacement stops the old one; - `doShutdown` shuts down the current and the replaced processor. - `DynamicRouterProducerTest`: both `process` methods fail with the `IllegalStateException` when the channel has no processor, and the callback is called once. Against main's `DynamicRouterEndpoint` compiled over the branch, all route tests fail. The stop/start test fails on its own (`... channel 'orders' already has a processor`); the others fail when their setup adds the route again. Mutants were also checked: - without the stop of the replaced processor, the other-options test fails; - stopping the old processor before creating the new one, or a `doShutdown` that does not shut the processors down, fails the component test; - shutting the pool of the replaced processor down fails the new `parallelProcessing` test; - reusing the processor whatever the URI fails the other-options and second-endpoint tests; - main's producer fails the 2 producer tests with a `NullPointerException`. The route test class ran 20 times in one JVM: 180 tests, 0 failures. All tests of camel-dynamic-router pass, including the ITs, which need no container: 169 tests (148 unit tests and the 21 ITs, run with surefire), 0 failures. The formatter and import-sort plugins ran, and the tree was clean afterwards. Stress (the red-team's plain-JVM probes, G7): a `toD` with `cacheSize=1`, alternating channels, 8 threads × 1,500 messages, default and `parallelProcessing=true`, gave 11,984 and 11,991 OK. The rest (16 and 9) failed with the core `ProducerCache` error `No producer, this processor has not been started` (eviction race in `ServicePool`, not in this module). There were no hangs and no errors from the Dynamic Router. With queued `parallelProcessing` exchanges and the endpoint removed or evicted, all exchanges completed. The RecipientList threads stay at one pool per channel over 50 remove/add cycles. Replacing the processor for a route that is added again with other options, while another route still has exchanges in flight in it, was checked with a probe: 10 running and 5 queued `parallelProcessing` exchanges, plus 10 more sent afterwards, all 25 completed. A `toD` that keeps sending to the old URI after the replacement fails with `ResolveEndpointFailedException`, as two URIs for one channel did before. Limits: every replacement keeps the replaced processor, stopped, until the component shuts down. Its pool threads (default profile) end after the 60 s keep-alive. Replacements come from a route added again with other options. They also come from a `toD` that puts different options for the same channel in its URI per message (for example `recipientMode=${header.mode}`) and whose producer cache evicts the endpoints. On main that `toD` fails ("only one processor per channel") whenever it has to create the endpoint again. With this change it routes, but it keeps one stopped processor per eviction. Follow-ups (I can open JIRAs if wanted): - Shut a replaced processor down once it has no exchange in flight, instead of keeping it until the component shuts down (needs an in-flight count on the send path, so it is left out of this fix). - Two first sends of a `toD` to a new channel with *different* URIs at the same time still race for the channel. With the same URI they now share the processor. - The core `ProducerCache` / `ServicePool` eviction race above. # 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-26453) filed for the change (usually before you start working on it): [CAMEL-26453](https://issues.apache.org/jira/browse/CAMEL-26453). # 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 camel-dynamic-router and ran its tests, including the formatter and import-sort plugins. I did not run the full root build. No option or generated file changed.) # 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]
