shashank created CAMEL-25018:
--------------------------------
Summary: ThrottlingInflightRoutePolicy and
ThrottlingExceptionRoutePolicy resume a consumer that the route controller
suspended: suspendRoute is not honoured and a graceful stop keeps consuming
until the timeout
Key: CAMEL-25018
URL: https://issues.apache.org/jira/browse/CAMEL-25018
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
Both throttling route policies resume the route's consumer with
{{RoutePolicySupport.resumeOrStartConsumer}} ->
{{ServiceHelper.resumeService}}, which resumes any suspended {{Suspendable}}
consumer, whoever suspended it:
* {{ThrottlingInflightRoutePolicy.throttle}} resumes on every
{{onExchangeDone}} where the route's inflight count is at or below the resume
threshold (70 % of {{maxInflightExchanges}}), i.e. on almost every completion
when the route is not busy;
* {{ThrottlingExceptionRoutePolicy.halfOpenCircuit}} / {{closeCircuit}} resume
from the half-open timer or from {{onExchangeDone}}.
The route controller uses the same consumer state: for
{{stopRoute}}/{{suspendRoute}} {{DefaultShutdownStrategy}} suspends a
{{Suspendable}} consumer first, then waits for the inflight exchanges, and only
then stops or suspends it for good. {{RoutePolicyAdvice}} skips
{{onExchangeDone}} only while the whole CamelContext is stopping, not while one
route is being stopped or suspended.
Consequences:
# {{ThrottlingInflightRoutePolicy}} + {{stopRoute}}/{{suspendRoute}} of a route
with inflight exchanges: the first exchange that completes resumes the
consumer, which takes new messages while the shutdown strategy waits for the
inflight count to reach 0. Under steady traffic the stop/suspend always runs
into the timeout and is then forced, failing the exchanges it just took in
(JMS/Kafka: redelivery or loss depending on the acknowledge mode).
# {{ThrottlingExceptionRoutePolicy}} + {{suspendRoute}} while the circuit is
open (an operator suspends the route during an outage): the half-open timer
resumes the consumer and the route consumes although its status is Suspended.
Batch consumers (file, FTP) are not affected by (1) because they refuse to
route while shutting down; timer, seda, JMS, Kafka and other {{Suspendable}}
consumers are.
*Reproduction*:
{noformat}
inflight-stop-timer: from("timer:t?period=100") with
ThrottlingInflightRoutePolicy (max 10); each exchange takes 900 ms,
the first one is held until stopRoute("r", 4 s) has suspended the consumer:
stopRoute(r, 4 s) in progress: consumer=Suspended, releasing the inflight
exchange
300 ms after the inflight exchange completed: consumer=Started
stopRoute returned after 4007 ms; exchanges started by the consumer after it
was suspended for the stop=5 (expected 0)
log: "Timeout occurred during graceful shutdown. Forcing the routes to be
shutdown now", then an exchange taken
during the stop fails with RejectedExecutionException
control (same route, a RoutePolicySupport that does nothing): consumer stays
Suspended, stopRoute returns after 1007 ms, 0 started
usersuspend: file consumer, ThrottlingExceptionRoutePolicy(1, 60000, 500,
null); a failing file opens the circuit, then suspendRoute("r"):
suspendRoute(r) returned: route=Suspended consumer=Suspended circuit=opened
2 s later: route=Suspended consumer=Started circuit=half opened, files
processed while the route is suspended=3/3
{noformat}
TLA+ models of both policies (the inflight policy's throttle split into its
steps, the controller's suspend/stop, the half-open timer) violate "the policy
never resumes a consumer the controller suspended"; with the fix below it
holds, and the policy still never leaves a consumer it suspended itself
suspended forever.
*Proposed fix:*
* the policies only resume a consumer they suspended themselves, and only
record that for a consumer that was started ({{ServiceHelper.suspendService}}
also returns true for an already stopped consumer, which would otherwise be
claimed and later restarted);
* they do not resume or start the consumer while the route is Suspending,
Suspended, Stopping or Stopped, or the CamelContext is stopping (a shared
{{RoutePolicySupport}} helper);
* the exception policy only resumes from OPEN (a HALF_OPEN close leaves the
consumer to the route controller).
Known remaining gap: the route status only changes after the graceful wait, so
a consumer the policy suspended itself before the stop began can still be
resumed during the wait. Closing that would need the shutdown strategy to
notify the route policies before it waits.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
