This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a4f2532b4262 CAMEL-25848: camel-quartz - a scheduled resume or start
starts a consumer that the scheduled suspend stopped (#28321)
a4f2532b4262 is described below
commit a4f2532b426202cbd63f0eff233cd7631fb9de01
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 22:04:13 2026 +0530
CAMEL-25848: camel-quartz - a scheduled resume or start starts a consumer
that the scheduled suspend stopped (#28321)
* CAMEL-25848: camel-quartz - a scheduled resume or start starts a consumer
that the scheduled suspend stopped
The scheduled route policies (CronScheduledRoutePolicy and
SimpleScheduledRoutePolicy) suspend the consumer of the route with
suspendOrStopConsumer, which stops a consumer that is not Suspendable
(the consumers of netty, mina, mllp, paho, paho-mqtt5, grpc,
azure-servicebus, google-pubsub and others). The route stays Started.
The scheduled resume only resumed a consumer that was suspended, and
logged "The Consumer ... is not suspended and cannot be resumed" for
a stopped one; the scheduled start of a Started route did nothing.
Such a consumer therefore stayed stopped after the first scheduled
suspend, until the route was restarted.
A scheduled resume or start of a Started route now also starts a
consumer that the scheduled suspend of the policy stopped (the policy
remembers it per route until the consumer or the route starts again or
the route stops), with resumeOrStartConsumer as the throttling route
policies do. A consumer that something else stopped is left stopped:
when a route 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
between must not start the consumer again. The policy also checks
isResumeOrStartConsumerAllowed first, as the throttling route policies
do.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25848: camel-quartz - the scheduled suspend records only a consumer
that it stopped
Review of #28321: suspendOrStopConsumer stops a non-Suspendable consumer
even when it is
already stopped, so the scheduled suspend also recorded a consumer that
another policy or the
shutdown strategy had stopped, and a later scheduled resume or start
started it again. The
SUSPEND action now checks the state before it suspends and records the
route only when this
suspend stopped the consumer.
The record is also cleared in onResume. The route controller calls onResume
only for a route
whose consumer is Suspendable (which the scheduled suspend suspends and
never records); a route
whose consumer is not Suspendable is stopped and started instead, which
clears the record in
onStop and onStart.
Tests (ScheduledRoutePolicyNonSuspendableConsumerTest):
- scheduledSuspendDoesNotRecordAConsumerThatWasAlreadyStopped fails when
the suspend records a
consumer that was already stopped;
- routeSuspendedAndResumedAfterTheScheduledSuspendKeepsNoRecordOfIt (route
controller suspend
and resume after the scheduled suspend, then another stop of the consumer
and a scheduled
resume) passes with and without the onResume line, as onStop/onStart
clear the record.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../routepolicy/quartz/ScheduledRoutePolicy.java | 76 +++++-
...duledRoutePolicyNonSuspendableConsumerTest.java | 271 +++++++++++++++++++++
2 files changed, 342 insertions(+), 5 deletions(-)
diff --git
a/components/camel-quartz/src/main/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicy.java
b/components/camel-quartz/src/main/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicy.java
index 1e8a7d1cb25c..5e6c61f8ac0c 100644
---
a/components/camel-quartz/src/main/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicy.java
+++
b/components/camel-quartz/src/main/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicy.java
@@ -18,8 +18,11 @@ package org.apache.camel.routepolicy.quartz;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import org.apache.camel.Consumer;
import org.apache.camel.NonManagedService;
import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
@@ -58,6 +61,9 @@ public abstract class ScheduledRoutePolicy extends
RoutePolicySupport
@Metadata(description = "Timeout (in millis) when stopping routes.",
defaultValue = "10000")
private int routeStopGracePeriod;
private TimeUnit timeUnit;
+ // the routes whose consumer the scheduled suspend stopped (it is not
Suspendable): only such a consumer is started
+ // again by a scheduled resume or start, not one that the shutdown of the
route, or another policy, stopped
+ private final Set<String> consumersStoppedBySuspend =
ConcurrentHashMap.newKeySet();
protected abstract Trigger createTrigger(Action action, Route route)
throws Exception;
@@ -69,8 +75,8 @@ public abstract class ScheduledRoutePolicy extends
RoutePolicySupport
if (routeStatus == ServiceStatus.Stopped) {
startRoute(route);
// here we just check the states of the Consumer
- } else if (ServiceHelper.isSuspended(route.getConsumer())) {
- resumeOrStartConsumer(route.getConsumer());
+ } else if (isSuspendedOrStoppedBySuspend(route)) {
+ resumeOrStartSuspendedConsumer(route);
}
} else if (action == Action.STOP) {
if (routeStatus == ServiceStatus.Started || routeStatus ==
ServiceStatus.Suspended) {
@@ -81,14 +87,21 @@ public abstract class ScheduledRoutePolicy extends
RoutePolicySupport
}
} else if (action == Action.SUSPEND) {
if (routeStatus == ServiceStatus.Started) {
- suspendOrStopConsumer(route.getConsumer());
+ Consumer consumer = route.getConsumer();
+ // only a consumer that this suspend stopped, not one that was
stopped already (by another policy, or
+ // by the shutdown strategy while the route is being stopped)
+ boolean alreadyStopped = ServiceHelper.isStopped(consumer);
+ suspendOrStopConsumer(consumer);
+ if (!alreadyStopped && ServiceHelper.isStopped(consumer)) {
+ consumersStoppedBySuspend.add(route.getId());
+ }
} else {
LOG.warn("Route is not in a started state and cannot be
suspended. The current route state is {}", routeStatus);
}
} else if (action == Action.RESUME) {
if (routeStatus == ServiceStatus.Started) {
- if (ServiceHelper.isSuspended(route.getConsumer())) {
- resumeOrStartConsumer(route.getConsumer());
+ if (isSuspendedOrStoppedBySuspend(route)) {
+ resumeOrStartSuspendedConsumer(route);
} else {
LOG.warn("The Consumer {} is not suspended and cannot be
resumed.", route.getConsumer());
}
@@ -98,6 +111,59 @@ public abstract class ScheduledRoutePolicy extends
RoutePolicySupport
}
}
+ /**
+ * Whether the consumer of a started route was suspended, or stopped by
the scheduled suspend as it is not
+ * {@link org.apache.camel.Suspendable} (see {@link
#suspendOrStopConsumer(Consumer)}), so a resume or start must
+ * resume or start it again. A consumer that something else stopped, such
as the shutdown strategy while the route
+ * is being stopped (the route status stays Started until the in-flight
exchanges are done), is left stopped.
+ */
+ private boolean isSuspendedOrStoppedBySuspend(Route route) {
+ Consumer consumer = route.getConsumer();
+ if (ServiceHelper.isSuspended(consumer)) {
+ return true;
+ }
+ if (ServiceHelper.isStopped(consumer)) {
+ return consumersStoppedBySuspend.contains(route.getId());
+ }
+ return false;
+ }
+
+ /**
+ * Resumes or starts the consumer of a started route, unless Camel is
stopping: the shutdown suspends or stops the
+ * consumers of the routes before it stops the routes, so a scheduled
resume or start that fires meanwhile must not
+ * start the consumer again (as the throttling route policies do, see
+ * {@link #isResumeOrStartConsumerAllowed(Route)}).
+ */
+ private void resumeOrStartSuspendedConsumer(Route route) throws Exception {
+ if (isResumeOrStartConsumerAllowed(route)) {
+ consumersStoppedBySuspend.remove(route.getId());
+ resumeOrStartConsumer(route.getConsumer());
+ } else {
+ LOG.debug("The consumer of route {} is not resumed or started as
Camel or the route is stopping", route.getId());
+ }
+ }
+
+ @Override
+ public void onStart(Route route) {
+ super.onStart(route);
+ // the route (and its consumer) started again
+ consumersStoppedBySuspend.remove(route.getId());
+ }
+
+ @Override
+ public void onStop(Route route) {
+ super.onStop(route);
+ consumersStoppedBySuspend.remove(route.getId());
+ }
+
+ @Override
+ public void onResume(Route route) {
+ super.onResume(route);
+ // the route (and its consumer) resumed; a route whose consumer is not
Suspendable is stopped and started by the
+ // route controller instead (onStop and onStart), so this only makes
sure that no record outlives a resume
+ consumersStoppedBySuspend.remove(route.getId());
+ }
+
@Override
public void onRemove(Route route) {
try {
diff --git
a/components/camel-quartz/src/test/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicyNonSuspendableConsumerTest.java
b/components/camel-quartz/src/test/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicyNonSuspendableConsumerTest.java
new file mode 100644
index 000000000000..6cf58f4a9e48
--- /dev/null
+++
b/components/camel-quartz/src/test/java/org/apache/camel/routepolicy/quartz/ScheduledRoutePolicyNonSuspendableConsumerTest.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.routepolicy.quartz;
+
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.Route;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.Suspendable;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.quartz.QuartzComponent;
+import org.apache.camel.impl.engine.DefaultShutdownStrategy;
+import org.apache.camel.spi.RoutePolicy;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.service.ServiceHelper;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A consumer that is not {@link Suspendable} is stopped when a scheduled
route policy suspends its route. The scheduled
+ * resume (or start) must start it again, as {@link
org.apache.camel.support.RoutePolicySupport#resumeOrStartConsumer}
+ * does.
+ */
+class ScheduledRoutePolicyNonSuspendableConsumerTest extends NoBuilderTest {
+
+ @Test
+ void scheduledResumeStartsConsumerThatWasStoppedBySuspend() throws
Exception {
+ context.addEndpoint("listener:in", new ListenerEndpoint());
+ context.getComponent("quartz", QuartzComponent.class)
+
.setPropertiesFile("org/apache/camel/routepolicy/quartz/myquartz.properties");
+ context.addRoutes(new RouteBuilder() {
+ public void configure() {
+ SimpleScheduledRoutePolicy policy = new
SimpleScheduledRoutePolicy();
+ long now = System.currentTimeMillis();
+ policy.setRouteSuspendDate(new Date(now + 500));
+ policy.setRouteSuspendRepeatCount(0);
+ policy.setRouteResumeDate(new Date(now + 2000));
+ policy.setRouteResumeRepeatCount(0);
+
+
from("listener:in").routeId("test").routePolicy(policy).to("mock:result");
+ }
+ });
+ context.start();
+
+ Consumer consumer = context.getRoute("test").getConsumer();
+
+ // the suspend stops the consumer, as it cannot be suspended
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertTrue(ServiceHelper.isStopped(consumer), "consumer should be stopped"));
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("test"));
+
+ // the resume must start it again
+ Awaitility.await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertTrue(ServiceHelper.isStarted(consumer), "consumer should be started
again"));
+ }
+
+ @Test
+ void scheduledStartStartsConsumerThatWasStoppedBySuspend() throws
Exception {
+ context.addEndpoint("listener:in", new ListenerEndpoint());
+ context.addRoutes(new RouteBuilder() {
+ public void configure() {
+ CronScheduledRoutePolicy policy = new
CronScheduledRoutePolicy();
+ // never fires by itself, the actions are run below as the
scheduler would
+ policy.setRouteSuspendTime("0 0 0 1 1 ? 2099");
+ policy.setRouteStartTime("0 0 0 1 1 ? 2099");
+
+
from("listener:in").routeId("test").routePolicy(policy).to("mock:result");
+ }
+ });
+ context.start();
+
+ Route route = context.getRoute("test");
+ ScheduledRoutePolicy policy = scheduledRoutePolicy(route);
+ Consumer consumer = route.getConsumer();
+
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.SUSPEND,
route);
+ assertTrue(ServiceHelper.isStopped(consumer), "consumer should be
stopped");
+ assertFalse(ServiceHelper.isSuspended(consumer));
+
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.START, route);
+ assertTrue(ServiceHelper.isStarted(consumer), "consumer should be
started again");
+ }
+
+ @Test
+ void scheduledSuspendDoesNotRecordAConsumerThatWasAlreadyStopped() throws
Exception {
+ Route route = startRouteWithActionsRunByTheTest();
+ ScheduledRoutePolicy policy = scheduledRoutePolicy(route);
+ Consumer consumer = route.getConsumer();
+
+ // something else stopped the consumer (such as a throttling policy),
and the route is still started
+ ServiceHelper.stopService(consumer);
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.SUSPEND,
route);
+
+ // the consumer was not stopped by the scheduled suspend, so the
scheduled resume or start leaves it stopped
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.RESUME,
route);
+ assertTrue(ServiceHelper.isStopped(consumer), "the scheduled resume
started a consumer it did not stop");
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.START, route);
+ assertTrue(ServiceHelper.isStopped(consumer), "the scheduled start
started a consumer it did not stop");
+ }
+
+ @Test
+ void routeSuspendedAndResumedAfterTheScheduledSuspendKeepsNoRecordOfIt()
throws Exception {
+ Route route = startRouteWithActionsRunByTheTest();
+ ScheduledRoutePolicy policy = scheduledRoutePolicy(route);
+ Consumer consumer = route.getConsumer();
+
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.SUSPEND,
route);
+ assertTrue(ServiceHelper.isStopped(consumer), "consumer should be
stopped");
+
+ // the route controller suspends and resumes the route: its consumer
is not Suspendable, so the route does not
+ // support suspension, and the route controller stops and starts the
route instead
+ context.getRouteController().suspendRoute("test");
+ context.getRouteController().resumeRoute("test");
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("test"));
+ route = context.getRoute("test");
+ Consumer started = route.getConsumer();
+ assertTrue(ServiceHelper.isStarted(started), "consumer should be
started by the route controller");
+
+ // something else stops the consumer: the scheduled suspend before the
resume of the route no longer counts
+ ServiceHelper.stopService(started);
+ policy.onJobExecute(ScheduledRoutePolicyConstants.Action.RESUME,
route);
+ assertTrue(ServiceHelper.isStopped(started), "the scheduled resume
started a consumer it did not stop");
+ }
+
+ /**
+ * Starts a route with a non-Suspendable consumer and a scheduled route
policy whose triggers never fire: the test
+ * runs the actions as the scheduler would.
+ */
+ private Route startRouteWithActionsRunByTheTest() throws Exception {
+ context.addEndpoint("listener:in", new ListenerEndpoint());
+ context.addRoutes(new RouteBuilder() {
+ public void configure() {
+ CronScheduledRoutePolicy policy = new
CronScheduledRoutePolicy();
+ policy.setRouteSuspendTime("0 0 0 1 1 ? 2099");
+ policy.setRouteResumeTime("0 0 0 1 1 ? 2099");
+ policy.setRouteStartTime("0 0 0 1 1 ? 2099");
+
+
from("listener:in").routeId("test").routePolicy(policy).to("mock:result");
+ }
+ });
+ context.start();
+ return context.getRoute("test");
+ }
+
+ @Test
+ void scheduledResumeOrStartDoesNotStartTheConsumerWhileCamelStops() throws
Exception {
+ List<String> startedDuringShutdown =
runScheduledResumeAndStartWhileStopping(false);
+
+ assertEquals(List.of(), startedDuringShutdown, "the consumer was
started again while Camel stopped");
+ }
+
+ @Test
+ void scheduledResumeOrStartDoesNotStartTheConsumerWhileTheRouteStops()
throws Exception {
+ // the route status stays Started while the shutdown strategy waits
for the in-flight exchanges
+ List<String> startedDuringShutdown =
runScheduledResumeAndStartWhileStopping(true);
+
+ assertEquals(List.of(), startedDuringShutdown, "the consumer was
started again while the route stopped");
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("test"));
+ }
+
+ /**
+ * Stops the route, or Camel, and runs the scheduled resume and start
after the shutdown strategy stopped the
+ * consumer, as a trigger that fires during the shutdown would.
+ *
+ * @return the actions after which the consumer was started
+ */
+ private List<String> runScheduledResumeAndStartWhileStopping(boolean
routeOnly) throws Exception {
+ context.addEndpoint("listener:in", new ListenerEndpoint());
+ List<String> startedDuringShutdown = new CopyOnWriteArrayList<>();
+ // the shutdown stops the consumer of the started route first (it is
not Suspendable), and stops the route
+ // later: run the scheduled actions in between, as a trigger that
fires during the shutdown would
+ context.setShutdownStrategy(new DefaultShutdownStrategy(context) {
+ @Override
+ protected void shutdownNow(String routeId, Consumer consumer) {
+ super.shutdownNow(routeId, consumer);
+ Route route = context.getRoute(routeId);
+ for (ScheduledRoutePolicyConstants.Action action : List.of(
+ ScheduledRoutePolicyConstants.Action.RESUME,
ScheduledRoutePolicyConstants.Action.START)) {
+ try {
+ scheduledRoutePolicy(route).onJobExecute(action,
route);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ if (ServiceHelper.isStarted(consumer)) {
+ startedDuringShutdown.add(action.name());
+ }
+ }
+ }
+ });
+ context.addRoutes(new RouteBuilder() {
+ public void configure() {
+ CronScheduledRoutePolicy policy = new
CronScheduledRoutePolicy();
+ // never fires by itself, the actions are run by the shutdown
strategy above
+ policy.setRouteResumeTime("0 0 0 1 1 ? 2099");
+ policy.setRouteStartTime("0 0 0 1 1 ? 2099");
+
+
from("listener:in").routeId("test").routePolicy(policy).to("mock:result");
+ }
+ });
+ context.start();
+ Consumer consumer = context.getRoute("test").getConsumer();
+ assertTrue(ServiceHelper.isStarted(consumer));
+
+ if (routeOnly) {
+ context.getRouteController().stopRoute("test");
+ } else {
+ context.stop();
+ }
+
+ assertTrue(ServiceHelper.isStopped(consumer), "consumer should be
stopped");
+ return startedDuringShutdown;
+ }
+
+ private static ScheduledRoutePolicy scheduledRoutePolicy(Route route) {
+ for (RoutePolicy policy : route.getRoutePolicyList()) {
+ if (policy instanceof ScheduledRoutePolicy srp) {
+ return srp;
+ }
+ }
+ throw new IllegalStateException("No scheduled route policy on route "
+ route.getId());
+ }
+
+ /**
+ * An endpoint whose consumer is not {@link Suspendable}, like the
consumers of the netty, mina, mllp or paho
+ * components.
+ */
+ private static final class ListenerEndpoint extends DefaultEndpoint {
+
+ ListenerEndpoint() {
+ super("listener:in", null);
+ }
+
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ Consumer answer = new DefaultConsumer(this, processor) {
+ };
+ configureConsumer(answer);
+ return answer;
+ }
+ }
+}
