allthingssecurity opened a new pull request, #27338: URL: https://github.com/apache/camel/pull/27338
# Description [CAMEL-25297](https://issues.apache.org/jira/browse/CAMEL-25297) `DirectProducer` (and `KameletProducer`, which has the same code) caches the consumer of its endpoint in two fields, `stateCounter` and `consumer`, which all threads sending with the producer share. When the consumer changes (its route is suspended or stopped, which removes the consumer from `DirectComponent` and increments the component's counter), the first exchange sets `stateCounter` to the new value and then looks up the consumer. With `block=true` (the default) that lookup waits up to `timeout` for the consumer to come back. While it waits, every other exchange sent with the same producer sees the new `stateCounter` together with the old `consumer` and skips the lookup. With a suspended route they are processed by that route, so suspending a direct route (route controller, JMX, `ThrottlingInflightRoutePolicy`) does not hold back concurrent senders: only the first exchange waits as documented. With a stopped route they fail at once with a `RejectedExecutionException` from its error handler instead of waiting for the route to start again. The cache was introduced by CAMEL-15690 (3.7.0) to avoid a lookup under the lock per exchange; the fast path stays as it is. This change keeps the consumer and the counter in one immutable holder (`record CachedConsumer(DirectConsumer consumer, int stateCounter)`, in a `volatile` field), in both producers. The counter is read before the lookup and stored with its result, so another exchange either sees the old holder (counter differs: it looks up the consumer too and waits) or the result of the lookup. The fast path is unchanged (a volatile read of the holder and of the counter); a holder is only allocated when the consumer changed. The defect was found with a TLA+ model of the cache: with two sending threads, "an exchange whose check starts after the consumer was removed is not sent to that consumer" is violated (thread 1 sets the counter and starts the lookup, the consumer is removed, thread 2's check passes with the old consumer). It holds with one thread, and with the holder for 2 and 3 threads. No upgrade guide entry: the change restores the documented behaviour of `block` for a suspended or stopped consumer (as before 3.7.0). Tests: new `DirectProducerSuspendedConsumerTest` (camel-core). Route `b` (`from("direct:b")`) is suspended after a first exchange warmed up the producer of `to("direct:b?timeout=20000")`. A `DirectComponent` subclass counts down a latch in `getConsumer`, so the test knows when the first exchange waits for the consumer; then a second exchange is sent. It must wait too (no message reaches the suspended route), and after `resumeRoute("b")` both are delivered. Without the main-code change: ``` DirectProducerSuspendedConsumerTest.testExchangesWaitForSuspendedConsumer No exchange should reach the suspended route ==> expected: <1> but was: <2> ``` The same test with `stopRoute("b")`/`startRoute("b")`; without the main-code change the second exchange fails at once: ``` DirectProducerSuspendedConsumerTest.testExchangesWaitForStoppedConsumer The second exchange should wait for the consumer of route b ==> expected: <false> but was: <true> ``` New `KameletProducerSuspendedConsumerTest` (camel-kamelet) does the same with a kamelet route (`kamelet:echo/echo?timeout=20000`, route template `kamelet:source -> mock:kamelet`) and a `KameletComponent` subclass; on main both tests fail the same way. With the change, `Direct*`, `*Suspend*` and `*Resume*` tests in camel-core pass (60 tests, 0 failures), and the `Kamelet*Test` tests of camel-kamelet pass (84 tests, 0 failures, 3 skipped). # 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). # 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 the affected modules, including the formatter and import-sort plugins. 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]
