allthingssecurity opened a new pull request, #26883: URL: https://github.com/apache/camel/pull/26883
# Description [CAMEL-25018](https://issues.apache.org/jira/browse/CAMEL-25018) `ThrottlingInflightRoutePolicy` and `ThrottlingExceptionRoutePolicy` resumed the route's consumer with `RoutePolicySupport.resumeOrStartConsumer`. That resumes any suspended `Suspendable` consumer, and starts any consumer that is not `Suspendable`, no matter who suspended or stopped it. The route controller uses the same consumer state. For `stopRoute`/`suspendRoute`, `DefaultShutdownStrategy` first suspends a `Suspendable` consumer, then waits for the route's inflight exchanges, and only then stops or suspends the route. `RoutePolicyAdvice` skips `onExchangeDone` only while the whole CamelContext is stopping, not while a single route is being stopped or suspended. What went wrong: 1. `ThrottlingInflightRoutePolicy` resumed the consumer whenever an exchange completed with the route's inflight count at or below the resume threshold (70% of `maxInflightExchanges`), so on almost every completion when the route was not busy. The first exchange that completed during a graceful `stopRoute`/`suspendRoute` resumed the consumer. The consumer then took new messages while the shutdown strategy waited for the inflight count to reach 0. Under steady traffic the stop always ran into the timeout and was forced, failing the exchanges it had just taken in. Timer, seda, JMS, Kafka and other `Suspendable` consumers are affected. Batch consumers (file, FTP) are not, because `isBatchAllowed` refuses to route while shutting down. 2. `ThrottlingExceptionRoutePolicy` resumed the consumer from the half open timer also when the route had been suspended in the meantime, for example by an operator during an outage. The route's status was Suspended, but it consumed. This change: both policies only resume a consumer they suspended themselves. They also do not resume when the route controller has suspended or stopped the route, or is suspending or stopping it, or when Camel is stopping. - The route and context check is a small shared helper, `RoutePolicySupport.isResumeOrStartConsumerAllowed(Route)`. The existing `resumeOrStartConsumer`/`suspendOrStopConsumer` are unchanged, so other policies (e.g. camel-quartz's `ScheduledRoutePolicy`) are not affected. - `ThrottlingInflightRoutePolicy` remembers the consumers it suspended. It forgets them in `onStart`/`onStop`/`onSuspend`/`onResume`, when the route controller takes over the consumer. It only suspends, and so remembers, a consumer that is started. `ServiceHelper.suspendService` also returns `true` for a consumer that is already stopped (it suspends a stopped `Suspendable` consumer, and stops a non-`Suspendable` one again), so a consumer that the shutdown strategy stopped could otherwise be claimed by the policy and restarted later. - `ThrottlingExceptionRoutePolicy` only resumes the consumer when the circuit is open, because the policy suspended the consumer when it opened the circuit. When half open, the consumer has already been resumed, so `closeCircuit` no longer resumes a consumer that the route controller suspended in the meantime. When a resume is not allowed, the circuit state still changes (half open or closed), and the consumer is left to the route controller. After `resumeRoute`, the next exchange closes or re-opens the circuit as usual. This is independent of the `openCircuit` fix for CAMEL-25014 (the circuit stays open after CAMEL-24903, apache/camel#26874). `git merge-tree` of the two branches merges cleanly, and `Throttl*RoutePolicy*` passes on the combined code (3 tests in camel-support, 14 in camel-core, 0 failures), including the tests of both changes. Behaviour changes to be aware of: - With a consumer that is not `Suspendable`, `ThrottlingInflightRoutePolicy` called `startService` on the consumer, and logged "resuming consumer", on every completion below the threshold. It now only starts a consumer it stopped itself. It also no longer restarts a consumer that the route controller stopped. - Known remaining gap: if the policy had suspended the consumer itself before a graceful `stopRoute`/`suspendRoute` begins, it can still resume it while the shutdown strategy waits for the inflight exchanges. The route's status only changes after that wait, and route policies are not told that a stop has begun. The shutdown strategy suspends or stops the consumer again after the wait, so the effect is limited to messages taken during the wait. Closing this would need a callback from `DefaultShutdownStrategy` to the route's policies before the wait, which I left out of this change. Tests: - `ThrottlingInflightRoutePolicyStopRouteTest`: a timer route with `ThrottlingInflightRoutePolicy`. The first exchange is held until `stopRoute("foo", 5 s)` has suspended the consumer, and then released. The stop must complete gracefully without taking new exchanges. Without the fix: ``` AssertionFailedError: No exchange should be started while the route is being stopped ==> expected: <0> but was: <1> ``` The stop ran into the 5 s timeout (6 s instead of 2 s). - `ThrottlingExceptionRoutePolicySuspendRouteTest`: a failure opens the circuit, then the route is suspended through the route controller. When the half open timer fires, the consumer must stay suspended. After `resumeRoute`, a good message closes the circuit. The circuit half opens after 3 s and the test waits up to 10 s, so the timer cannot fire before the test has checked that the circuit is open, also on a slow CI machine. Without the fix: ``` AssertionFailedError: The consumer of the suspended route should not be resumed ==> expected: <true> but was: <false> ``` - `ThrottlingInflightRoutePolicyStoppedConsumerTest`: the route's consumer is stopped by someone else while the route is still started, as the shutdown strategy does. Then an exchange completes with more exchanges than the maximum inflight, and later one completes with none inflight. The consumer must stay stopped. Without the started check (and without the fix): ``` AssertionFailedError: The policy should not resume a consumer that it did not suspend ==> expected: <true> but was: <false> ``` With the fix all pass. `*Throttl*,*RoutePolicy*,*Suspend*` in camel-support, camel-core and camel-management: 3 + 119 + 16 tests, 0 failures. Found with TLA+ models of both policies racing with the route controller, then reproduced against the real classes. # 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]
