[
https://issues.apache.org/jira/browse/CAMEL-25018?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25018:
--------------------------------
Fix Version/s: 4.23.0
> 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
> Priority: Minor
> Fix For: 4.23.0
>
>
> 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)
