[
https://issues.apache.org/jira/browse/CAMEL-25848?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25848.
---------------------------------
Fix Version/s: 4.24.0
Resolution: Fixed
> camel-quartz - CronScheduledRoutePolicy / SimpleScheduledRoutePolicy: a
> consumer that cannot be suspended (netty, mina, mllp, paho, ...) is stopped
> by the scheduled suspend and never started again by the scheduled resume
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25848
> URL: https://issues.apache.org/jira/browse/CAMEL-25848
> Project: Camel
> Issue Type: Bug
> Components: camel-quartz
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> {{ScheduledRoutePolicy.onJobExecute}} (camel-quartz, main 20cdb309f7d0, lines
> 64-99) is shared by {{CronScheduledRoutePolicy}} and
> {{SimpleScheduledRoutePolicy}}. The scheduled suspend calls
> {{suspendOrStopConsumer(route.getConsumer())}}, which suspends a
> {{Suspendable}} consumer and *stops* any other consumer
> ({{ServiceHelper.suspendService}}). The route itself stays {{Started}}.
> The scheduled resume and start only look for a suspended consumer:
> {code:java}
> } else if (action == Action.RESUME) {
> if (routeStatus == ServiceStatus.Started) {
> if (ServiceHelper.isSuspended(route.getConsumer())) {
> resumeOrStartConsumer(route.getConsumer());
> } else {
> LOG.warn("The Consumer {} is not suspended and cannot be
> resumed.", route.getConsumer());
> }
> {code}
> The START action of a {{Started}} route has the same {{isSuspended}} check. A
> stopped consumer is not suspended, so after the first scheduled suspend a
> consumer that is not {{Suspendable}} is never started again: every scheduled
> resume logs the WARN above, and the route stays {{Started}} without a
> consumer until someone restarts the route or the application.
> Consumers that are not {{Suspendable}} on main include those of camel-netty
> (TCP/UDP), camel-mina, camel-mllp, camel-paho, camel-paho-mqtt5, camel-grpc,
> camel-azure-servicebus, camel-azure-eventhubs, camel-google-pubsub,
> camel-aws2-kinesis (KCL), camel-debezium and camel-mongodb (tailable cursor,
> change streams). The page of the policy describes exactly this use:
> suspension stops the consumer "from listening on a given port" and resumption
> makes the route accept requests again. For example a nightly maintenance
> window ({{routeSuspendTime}} / {{routeResumeTime}}) on a netty or MLLP
> listener leaves the port closed for good after the first night.
> {{resumeOrStartConsumer}} itself already starts a stopped consumer that is
> not {{Suspendable}} (the throttling route policies rely on it); only the
> guard in front of it is too narrow.
> *Fix*: a scheduled resume, and a scheduled start of a {{Started}} route,
> resume or start the consumer when it is suspended *or was stopped by the
> scheduled suspend of the policy* (the policy remembers that per route until
> the consumer or the route starts again, or the route stops). A consumer that
> something else stopped is left stopped: when a route ({{stopRoute}}) or Camel
> stops, the shutdown strategy stops a consumer that is not {{Suspendable}} and
> then waits for the in-flight exchanges while the route status stays
> {{Started}}, so a trigger that fires in that window must not start the
> consumer again. The policy also checks
> {{isResumeOrStartConsumerAllowed(route)}} first, as the throttling route
> policies do. Nothing else changes: a stopped route is still started with
> {{startRoute}}, and a running consumer still gets the WARN on resume.
> *Tests*: new {{ScheduledRoutePolicyNonSuspendableConsumerTest}} (4 tests)
> with a test endpoint whose consumer is a plain {{DefaultConsumer}} (not
> {{Suspendable}}). {{scheduledResumeStartsConsumerThatWasStoppedBySuspend}}
> uses a {{SimpleScheduledRoutePolicy}} with a suspend date and a resume date
> and real Quartz triggers; on main the suspend stops the consumer and the
> resume never starts it ({{ConditionTimeoutException ... consumer should be
> started again ==> expected: <true> but was: <false> within 5 seconds}}).
> {{scheduledStartStartsConsumerThatWasStoppedBySuspend}} runs the SUSPEND and
> START actions of a {{CronScheduledRoutePolicy}} as the scheduler would; on
> main it fails with {{consumer should be started again ==> expected: <true>
> but was: <false>}}. Both pass with the fix.
> {{scheduledResumeOrStartDoesNotStartTheConsumerWhileCamelStops}} and
> {{scheduledResumeOrStartDoesNotStartTheConsumerWhileTheRouteStops}} run the
> RESUME and START actions from a {{DefaultShutdownStrategy}} subclass right
> after the shutdown stopped the consumer ({{context.stop()}}, {{stopRoute}});
> the consumer must stay stopped. Started for any suspended-or-stopped
> consumer, they fail with {{the consumer was started again while Camel stopped
> ==> expected: <[]> but was: <[RESUME, START]>}} (without the
> {{isResumeOrStartConsumerAllowed}} check) and {{the consumer was started
> again while the route stopped ==> expected: <[]> but was: <[RESUME, START]>}}
> (without the per-route record of the suspend). camel-quartz unit tests with
> the fix: 112 run, 0 failures.
> *How it was found*: a review of the route policies that suspend a consumer,
> comparing how each one undoes {{suspendOrStopConsumer}} (the throttling
> policies call {{resumeOrStartConsumer}} directly; the scheduled policies
> guard it with {{isSuspended}}). The JUnit tests above are the evidence; it
> was not run with a netty or MLLP route.
> *Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x
> (the same {{isSuspended}} checks in {{ScheduledRoutePolicy}}, lines 72 and
> 90, and {{NettyConsumer}} is not {{Suspendable}} on each; read, not run). The
> check on the START action came with CAMEL-5555 (start of a suspended route).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
