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 92c353446b56 CAMEL-26449: camel-core - DefaultShutdownStrategy: a
route stop that times out must not force or cut short the CamelContext stop
(#28805)
92c353446b56 is described below
commit 92c353446b561a1956e263489d113bbd5a7bbe38
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Nov 2 16:22:52 2026 +0530
CAMEL-26449: camel-core - DefaultShutdownStrategy: a route stop that times
out must not force or cut short the CamelContext stop (#28805)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/impl/engine/DefaultRoute.java | 5 +-
.../camel/impl/engine/DefaultShutdownStrategy.java | 85 +++--
...aultShutdownStrategyConcurrentShutdownTest.java | 420 +++++++++++++++++++++
3 files changed, 482 insertions(+), 28 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoute.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoute.java
index 03047b959c83..7156080c38a0 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoute.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoute.java
@@ -25,6 +25,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.camel.CamelContext;
import org.apache.camel.Channel;
@@ -114,7 +115,9 @@ public class DefaultRoute extends ServiceSupport implements
Route {
private final Endpoint endpoint;
private final Map<String, Object> properties = new HashMap<>();
- private final List<Service> services = new ArrayList<>();
+ // copy-on-write, as the shutdown task of a graceful shutdown iterates the
services of a route while another
+ // thread can shut the route down (a route stop that runs while the
CamelContext is stopping clears them).
+ private final List<Service> services = new CopyOnWriteArrayList<>();
private final List<Service> servicesToStop = new ArrayList<>();
// the services added with addService, which are kept when the services
are gathered again
private final List<Service> addedServices = new ArrayList<>();
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
index dc45a9770d47..67cbe7bff753 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultShutdownStrategy.java
@@ -34,6 +34,9 @@ import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
@@ -120,6 +123,8 @@ public class DefaultShutdownStrategy extends ServiceSupport
implements ShutdownS
private final CamelLogger logger = new CamelLogger(LOG,
LoggingLevel.DEBUG);
private CamelContext camelContext;
+ // guards the lazy creation of the executor, as several shutdowns can run
at the same time
+ private final Lock executorLock = new ReentrantLock();
private ExecutorService executor;
private long timeout = 45;
private TimeUnit timeUnit = TimeUnit.SECONDS;
@@ -130,7 +135,7 @@ public class DefaultShutdownStrategy extends ServiceSupport
implements ShutdownS
private boolean forceShutdown;
private final AtomicBoolean timeoutOccurred = new AtomicBoolean();
- private volatile Future<?> currentShutdownTaskFuture;
+ private final AtomicReference<Future<?>> currentShutdownTaskFuture = new
AtomicReference<>();
public DefaultShutdownStrategy() {
}
@@ -207,32 +212,43 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
logger.log(msg);
}
- // use another thread to perform the shutdowns so we can support
timeout
+ // use another thread to perform the shutdowns so we can support
timeout.
+ // The future and the timeout flag of this shutdown are local: several
shutdowns can run at the same time (a
+ // route is stopped or suspended while the CamelContext is stopping),
and the timeout of one must not cancel,
+ // force or cut short the graceful shutdown of another. The fields are
only set for the getters.
+ final AtomicBoolean shutdownTimeoutOccurred = new AtomicBoolean();
+ Future<?> future = null;
+ boolean interrupted = false;
timeoutOccurred.set(false);
try {
- currentShutdownTaskFuture = getExecutorService().submit(new
ShutdownTask(
+ future = getExecutorService().submit(new ShutdownTask(
context, routesOrdered, timeout, timeUnit, suspendOnly,
- abortAfterTimeout, timeoutOccurred,
isLogInflightExchangesOnTimeout()));
- currentShutdownTaskFuture.get(timeout, timeUnit);
+ abortAfterTimeout, shutdownTimeoutOccurred,
isLogInflightExchangesOnTimeout()));
+ currentShutdownTaskFuture.set(future);
+ future.get(timeout, timeUnit);
} catch (RejectedExecutionException e) {
// the task was rejected
} catch (ExecutionException e) {
// unwrap execution exception
throw
RuntimeCamelException.wrapRuntimeCamelException(e.getCause());
} catch (Exception e) {
- // either timeout or interrupted exception was thrown so this is
okay
- // as interrupted would mean cancel was called on the
currentShutdownTaskFuture to signal a forced timeout
+ // the timeout of this shutdown elapsed, or the calling thread was
interrupted (handled as a timeout)
+ interrupted = e instanceof InterruptedException;
// we hit a timeout, so set the flag
+ shutdownTimeoutOccurred.set(true);
timeoutOccurred.set(true);
// timeout then cancel the task
- if (currentShutdownTaskFuture != null) {
- currentShutdownTaskFuture.cancel(true);
+ if (future != null) {
+ future.cancel(true);
}
- // signal we are forcing shutdown now, since timeout occurred
- this.forceShutdown = forceShutdown;
+ // signal we are forcing shutdown now, since timeout occurred
(another shutdown that times out, and
+ // is not forced, must not turn it off again)
+ if (forceShutdown) {
+ this.forceShutdown = true;
+ }
// if set, stop processing and return false to indicate that the
shutdown is aborting
if (!forceShutdown && abortAfterTimeout) {
@@ -240,7 +256,7 @@ public class DefaultShutdownStrategy extends ServiceSupport
implements ShutdownS
+ " Notice: some resources may still be running as
graceful shutdown did not complete successfully.");
// we attempt to force shutdown so lets log the current
inflight exchanges which are affected
- logInflightExchanges(context, routes,
isLogInflightExchangesOnTimeout());
+ logInflightExchanges(context, routesOrdered,
isLogInflightExchangesOnTimeout());
return false;
} else {
@@ -249,13 +265,13 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
+ " Notice: some resources may still be running
as graceful shutdown did not complete successfully.");
// we attempt to force shutdown so lets log the current
inflight exchanges which are affected
- logInflightExchanges(context, routes,
isLogInflightExchangesOnTimeout());
+ logInflightExchanges(context, routesOrdered,
isLogInflightExchangesOnTimeout());
// force the routes to shutdown now
shutdownRoutesNow(routesOrdered);
// now the route consumers has been shutdown, then prepare
route services for shutdown now (forced)
- for (RouteStartupOrder order : routes) {
+ for (RouteStartupOrder order : routesOrdered) {
for (Service service : order.getServices()) {
prepareShutdown(service, false, true, true,
isSuppressLoggingOnTimeout());
}
@@ -264,11 +280,16 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
LOG.warn("Timeout occurred during graceful shutdown. Will
ignore shutting down the remainder routes."
+ " Notice: some resources may still be running
as graceful shutdown did not complete successfully.");
- logInflightExchanges(context, routes,
isLogInflightExchangesOnTimeout());
+ logInflightExchanges(context, routesOrdered,
isLogInflightExchangesOnTimeout());
}
}
} finally {
- currentShutdownTaskFuture = null;
+ // only clear the future of this shutdown, not the one of another
shutdown that runs meanwhile
+ currentShutdownTaskFuture.compareAndSet(future, null);
+ if (interrupted) {
+ // restore the interrupt flag, now that the routes have been
shutdown (forced) or the shutdown aborted
+ Thread.currentThread().interrupt();
+ }
}
if (logger.shouldLog()) {
@@ -377,12 +398,12 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
}
/**
- * Future for the current shutdown task, when a task is in progress.
+ * Future for the task of the shutdown that was started last, while that
shutdown is in progress, or <tt>null</tt>.
* <p/>
* Important: This API is only for advanced use-cases.
*/
public Future<?> getCurrentShutdownTaskFuture() {
- return currentShutdownTaskFuture;
+ return currentShutdownTaskFuture.get();
}
/**
@@ -464,11 +485,16 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
}
private ExecutorService getExecutorService() {
- if (executor == null) {
- // use a thread pool that allow to terminate idle threads so they
do not hang around forever
- executor =
camelContext.getExecutorServiceManager().newSingleThreadExecutor(this,
"ShutdownTask");
+ executorLock.lock();
+ try {
+ if (executor == null) {
+ // use a thread pool that allow to terminate idle threads so
they do not hang around forever
+ executor =
camelContext.getExecutorServiceManager().newSingleThreadExecutor(this,
"ShutdownTask");
+ }
+ return executor;
+ } finally {
+ executorLock.unlock();
}
- return executor;
}
@Override
@@ -481,11 +507,16 @@ public class DefaultShutdownStrategy extends
ServiceSupport implements ShutdownS
@Override
protected void doShutdown() throws Exception {
- if (executor != null) {
- // force shutting down as we are shutting down Camel
- camelContext.getExecutorServiceManager().shutdownNow(executor);
- // should clear executor so we can restart by creating a new
thread pool
- executor = null;
+ executorLock.lock();
+ try {
+ if (executor != null) {
+ // force shutting down as we are shutting down Camel
+ camelContext.getExecutorServiceManager().shutdownNow(executor);
+ // should clear executor so we can restart by creating a new
thread pool
+ executor = null;
+ }
+ } finally {
+ executorLock.unlock();
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultShutdownStrategyConcurrentShutdownTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultShutdownStrategyConcurrentShutdownTest.java
new file mode 100644
index 000000000000..787e761d9bbc
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/DefaultShutdownStrategyConcurrentShutdownTest.java
@@ -0,0 +1,420 @@
+/*
+ * 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.impl.engine;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.StatefulService;
+import org.apache.camel.Suspendable;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.ShutdownAware;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.service.ServiceSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * A route is stopped (with a short timeout) while the CamelContext is stopped
(with a long timeout), in either order.
+ * The timeout of the route stop must not cancel, force or cut short the
graceful shutdown of the CamelContext: the
+ * CamelContext must wait for the exchange that is still in flight on another
route, and not stop its consumer under it.
+ * Also when the route stop shuts its route down while the CamelContext stop
iterates the services of that route. And a
+ * route stop that times out after a forced CamelContext stop must not turn
the forced shutdown off.
+ */
+class DefaultShutdownStrategyConcurrentShutdownTest extends ContextTestSupport
{
+
+ private final CountDownLatch releaseA = new CountDownLatch(1);
+ private final CountDownLatch releaseB = new CountDownLatch(1);
+ // holds the task of the CamelContext stop while it iterates the services
of route a
+ private final AtomicBoolean holdPendingCheck = new AtomicBoolean();
+ private final CountDownLatch inPendingCheck = new CountDownLatch(1);
+ private final CountDownLatch releasePendingCheck = new CountDownLatch(1);
+ // holds the CamelContext stop while it prepares route a for a forced
shutdown
+ private final AtomicBoolean holdForcedPrepare = new AtomicBoolean();
+ private final CountDownLatch inForcedPrepare = new CountDownLatch(1);
+ private final CountDownLatch releaseForcedPrepare = new CountDownLatch(1);
+
+ @AfterEach
+ void releaseExchanges() {
+ releaseA.countDown();
+ releaseB.countDown();
+ releasePendingCheck.countDown();
+ releaseForcedPrepare.countDown();
+ }
+
+ @Test
+ void testRouteStopTimeoutDoesNotForceContextStop() throws Exception {
+ DefaultShutdownStrategy strategy = (DefaultShutdownStrategy)
context.getShutdownStrategy();
+ strategy.setTimeout(60);
+
+ StatefulService consumerB = (StatefulService)
context.getRoute("b").getConsumer();
+
+ // an exchange in flight on each route
+ CompletableFuture<Exchange> a = consume("a", "A");
+ CompletableFuture<Exchange> b = consume("b", "B");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size("a") == 1
+ && context.getInflightRepository().size("b") == 1);
+
+ // stop route a with a short timeout: its graceful shutdown waits for
the exchange on route a
+ AtomicReference<Throwable> stopRouteFailure = new AtomicReference<>();
+ Thread stopRoute = new Thread(() -> {
+ try {
+ context.getRouteController().stopRoute("a", 5,
TimeUnit.SECONDS);
+ } catch (Throwable e) {
+ stopRouteFailure.set(e);
+ }
+ }, "stop-route-a");
+ stopRoute.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
strategy.getCurrentShutdownTaskFuture() != null);
+ Future<?> routeTask = strategy.getCurrentShutdownTaskFuture();
+
+ // meanwhile the CamelContext is stopped (60 seconds timeout)
+ AtomicReference<Throwable> stopContextFailure = new
AtomicReference<>();
+ Thread stopContext = new Thread(() -> {
+ try {
+ context.stop();
+ } catch (Throwable e) {
+ stopContextFailure.set(e);
+ }
+ }, "stop-context");
+ stopContext.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
strategy.getCurrentShutdownTaskFuture() != null
+ && strategy.getCurrentShutdownTaskFuture() != routeTask);
+ // the CamelContext stop has started before the route stop times out
+ assertTrue(stopRoute.isAlive(), "the route stop timed out before the
CamelContext stop started");
+
+ // the route stop times out (its exchange is still in flight) and
forces route a
+ stopRoute.join(20_000);
+ assertFalse(stopRoute.isAlive(), "the route stop did not complete");
+ assertNull(stopRouteFailure.get());
+ releaseA.countDown();
+
+ // the CamelContext stop waits for the exchange on route b: its
consumer is suspended, not stopped under it
+ await("the CamelContext stop suspends route b").atMost(20,
TimeUnit.SECONDS)
+ .until(() -> consumerB.getStatus() == ServiceStatus.Suspended
|| consumerB.isStopping()
+ || consumerB.isStopped());
+ if (consumerB.isStopping() || consumerB.isStopped()) {
+ fail("The consumer of route b was stopped while its exchange was
in flight (status "
+ + consumerB.getStatus() + ", force shutdown " +
strategy.isForceShutdown()
+ + "): the timeout of the route stop forced the graceful
shutdown of the CamelContext");
+ }
+ assertEquals(1, context.getInflightRepository().size("b"));
+
+ // once the exchange completes the CamelContext stops, gracefully
+ releaseB.countDown();
+ stopContext.join(30_000);
+ assertFalse(stopContext.isAlive(), "the CamelContext stop did not
complete");
+ assertNull(stopContextFailure.get());
+ assertTrue(context.isStopped());
+ Exchange exchangeB = b.get(10, TimeUnit.SECONDS);
+ assertNull(exchangeB.getException());
+ assertEquals("Done B", exchangeB.getMessage().getBody());
+ assertFalse(strategy.isForceShutdown());
+ assertTrue(a.isDone());
+ }
+
+ @Test
+ void testRouteStopTimeoutDoesNotCutShortContextStop() throws Exception {
+ DefaultShutdownStrategy strategy = (DefaultShutdownStrategy)
context.getShutdownStrategy();
+ strategy.setTimeout(60);
+ StatefulService consumerB = (StatefulService)
context.getRoute("b").getConsumer();
+
+ // an exchange in flight on each route
+ CompletableFuture<Exchange> a = consume("a", "A");
+ CompletableFuture<Exchange> b = consume("b", "B");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size("a") == 1
+ && context.getInflightRepository().size("b") == 1);
+
+ // the CamelContext is stopped (60 seconds timeout), and waits for the
exchanges in flight
+ AtomicReference<Throwable> stopContextFailure = new
AtomicReference<>();
+ Thread stopContext = new Thread(() -> {
+ try {
+ context.stop();
+ } catch (Throwable e) {
+ stopContextFailure.set(e);
+ }
+ }, "stop-context");
+ stopContext.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> consumerB.getStatus()
== ServiceStatus.Suspended);
+
+ // meanwhile route a is stopped with a short timeout, which times out
(its task waits for the one of the
+ // CamelContext stop)
+ AtomicReference<Throwable> stopRouteFailure = new AtomicReference<>();
+ Thread stopRoute = new Thread(() -> {
+ try {
+ context.getRouteController().stopRoute("a", 1,
TimeUnit.SECONDS);
+ } catch (Throwable e) {
+ stopRouteFailure.set(e);
+ }
+ }, "stop-route-a");
+ stopRoute.start();
+ stopRoute.join(20_000);
+ assertFalse(stopRoute.isAlive(), "the route stop did not complete");
+ assertNull(stopRouteFailure.get());
+ assertTrue(strategy.isTimeoutOccurred(), "the route stop did not time
out");
+
+ // the CamelContext stop keeps waiting for the exchange on route b
(its wait loop checks every second)
+ await("the CamelContext stop keeps waiting for route b").during(3,
TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> consumerB.getStatus() == ServiceStatus.Suspended
&& stopContext.isAlive());
+ assertEquals(1, context.getInflightRepository().size("b"));
+ assertFalse(strategy.isForceShutdown());
+
+ // once the exchanges complete the CamelContext stops, gracefully
+ releaseA.countDown();
+ releaseB.countDown();
+ stopContext.join(30_000);
+ assertFalse(stopContext.isAlive(), "the CamelContext stop did not
complete");
+ assertNull(stopContextFailure.get());
+ Exchange exchangeB = b.get(10, TimeUnit.SECONDS);
+ assertNull(exchangeB.getException());
+ assertEquals("Done B", exchangeB.getMessage().getBody());
+ assertTrue(a.isDone());
+ }
+
+ @Test
+ void testRouteShutdownDuringContextStopDoesNotFailItsWait() throws
Exception {
+ DefaultShutdownStrategy strategy = (DefaultShutdownStrategy)
context.getShutdownStrategy();
+ strategy.setTimeout(60);
+ StatefulService consumerB = (StatefulService)
context.getRoute("b").getConsumer();
+ CompletableFuture<Exchange> b = consume("b", "B");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size("b") == 1);
+
+ // the task of the CamelContext stop is held while it iterates the
services of route a (to count their pending
+ // exchanges)
+ holdPendingCheck.set(true);
+ AtomicReference<Throwable> stopContextFailure = new
AtomicReference<>();
+ Thread stopContext = new Thread(() -> {
+ try {
+ context.stop();
+ } catch (Throwable e) {
+ stopContextFailure.set(e);
+ }
+ }, "stop-context");
+ stopContext.start();
+ assertTrue(inPendingCheck.await(10, TimeUnit.SECONDS), "the
CamelContext stop did not check route a");
+
+ // meanwhile route a is stopped: its task waits behind the one of the
CamelContext stop, so it times out, and
+ // as the CamelContext is stopping it shuts route a down, which clears
the services of the route
+ AtomicReference<Throwable> stopRouteFailure = new AtomicReference<>();
+ Thread stopRoute = new Thread(() -> {
+ try {
+ context.getRouteController().stopRoute("a", 1,
TimeUnit.SECONDS);
+ } catch (Throwable e) {
+ stopRouteFailure.set(e);
+ }
+ }, "stop-route-a");
+ stopRoute.start();
+ stopRoute.join(20_000);
+ assertFalse(stopRoute.isAlive(), "the route stop did not complete");
+ assertNull(stopRouteFailure.get());
+ assertTrue(context.getRoute("a").getServices().isEmpty(), "the route
stop did not shut down route a");
+ assertFalse(strategy.isForceShutdown());
+
+ // the task of the CamelContext stop goes on iterating, and keeps
waiting for the exchange on route b
+ releasePendingCheck.countDown();
+ await("the CamelContext stop keeps waiting for route b").during(3,
TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> consumerB.getStatus() == ServiceStatus.Suspended
&& stopContext.isAlive());
+ assertEquals(1, context.getInflightRepository().size("b"));
+
+ releaseB.countDown();
+ stopContext.join(30_000);
+ assertFalse(stopContext.isAlive(), "the CamelContext stop did not
complete");
+ assertNull(stopContextFailure.get());
+ Exchange exchangeB = b.get(10, TimeUnit.SECONDS);
+ assertNull(exchangeB.getException());
+ assertEquals("Done B", exchangeB.getMessage().getBody());
+ }
+
+ @Test
+ void testRouteStopTimeoutKeepsForcedContextStop() throws Exception {
+ DefaultShutdownStrategy strategy = (DefaultShutdownStrategy)
context.getShutdownStrategy();
+ // the CamelContext stop times out (1 second); the route stop (60
seconds) is cut short by an interrupt, which is
+ // handled as a timeout, once the CamelContext stop is forcing the
routes: no race between the two timeouts
+ strategy.setTimeout(1);
+ holdForcedPrepare.set(true);
+
+ CompletableFuture<Exchange> a = consume("a", "A");
+ CompletableFuture<Exchange> b = consume("b", "B");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
context.getInflightRepository().size("a") == 1
+ && context.getInflightRepository().size("b") == 1);
+
+ AtomicReference<Throwable> stopRouteFailure = new AtomicReference<>();
+ AtomicBoolean stopRouteInterrupted = new AtomicBoolean();
+ Thread stopRoute = new Thread(() -> {
+ try {
+ context.getRouteController().stopRoute("a", 60,
TimeUnit.SECONDS);
+
stopRouteInterrupted.set(Thread.currentThread().isInterrupted());
+ } catch (Throwable e) {
+ stopRouteFailure.set(e);
+ }
+ }, "stop-route-a");
+ stopRoute.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
strategy.getCurrentShutdownTaskFuture() != null);
+
+ AtomicReference<Throwable> stopContextFailure = new
AtomicReference<>();
+ Thread stopContext = new Thread(() -> {
+ try {
+ context.stop();
+ } catch (Throwable e) {
+ stopContextFailure.set(e);
+ }
+ }, "stop-context");
+ stopContext.start();
+
+ // the CamelContext stop times out and forces the routes to shut down;
it is held while it prepares route a
+ assertTrue(inForcedPrepare.await(20, TimeUnit.SECONDS), "the
CamelContext stop did not force route a");
+ assertTrue(strategy.isForceShutdown());
+ assertTrue(stopRoute.isAlive(), "the route stop timed out before the
CamelContext stop");
+
+ // then the route stop times out (interrupted), which must not turn
the forced shutdown off again
+ stopRoute.interrupt();
+ stopRoute.join(20_000);
+ assertFalse(stopRoute.isAlive(), "the route stop did not complete");
+ assertNull(stopRouteFailure.get());
+ assertTrue(stopRouteInterrupted.get(), "the route stop did not restore
the interrupt flag");
+ assertTrue(strategy.isForceShutdown(), "the route stop turned off the
forced shutdown of the CamelContext");
+
+ releaseForcedPrepare.countDown();
+ releaseA.countDown();
+ releaseB.countDown();
+ stopContext.join(30_000);
+ assertFalse(stopContext.isAlive(), "the CamelContext stop did not
complete");
+ assertNull(stopContextFailure.get());
+ assertNotNull(a.get(10, TimeUnit.SECONDS));
+ assertNotNull(b.get(10, TimeUnit.SECONDS));
+ }
+
+ /**
+ * The consumer of the route receives a message (in another thread).
+ */
+ private CompletableFuture<Exchange> consume(String routeId, String body) {
+ DefaultConsumer consumer = (DefaultConsumer)
context.getRoute(routeId).getConsumer();
+ Exchange exchange = consumer.createExchange(false);
+ exchange.getMessage().setBody(body);
+ CompletableFuture<Exchange> answer = new CompletableFuture<>();
+ Thread t = new Thread(
+ () -> consumer.getAsyncProcessor().process(exchange, doneSync
-> answer.complete(exchange)),
+ "consumer-" + routeId);
+ t.start();
+ return answer;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a consumer that is suspended (not stopped) during a
graceful shutdown
+ context.addComponent("suspendable", new DefaultComponent() {
+ @Override
+ protected Endpoint createEndpoint(String uri, String
remaining, Map<String, Object> parameters) {
+ return new DefaultEndpoint(uri, this) {
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Consumer createConsumer(Processor
processor) {
+ return new SuspendableConsumer(this,
processor);
+ }
+ };
+ }
+ });
+
+ from("suspendable:a").routeId("a")
+ .process(new PendingCheck())
+ .process(e -> assertTrue(releaseA.await(60,
TimeUnit.SECONDS)))
+ .setBody(simple("Done ${body}"));
+ from("suspendable:b").routeId("b")
+ .process(e -> assertTrue(releaseB.await(60,
TimeUnit.SECONDS)))
+ .setBody(simple("Done ${body}"));
+ }
+ };
+ }
+
+ /**
+ * A service of route a: it can hold the task of the CamelContext stop
while that counts the pending exchanges of
+ * route a, or the CamelContext stop while that prepares route a for a
forced shutdown.
+ */
+ private final class PendingCheck extends ServiceSupport implements
Processor, ShutdownAware {
+ @Override
+ public void process(Exchange exchange) {
+ // noop
+ }
+
+ @Override
+ public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
+ return false;
+ }
+
+ @Override
+ public int getPendingExchangesSize() {
+ if (Thread.currentThread().getName().contains("ShutdownTask") &&
holdPendingCheck.compareAndSet(true, false)) {
+ inPendingCheck.countDown();
+ try {
+ assertTrue(releasePendingCheck.await(30,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return 0;
+ }
+
+ @Override
+ public void prepareShutdown(boolean suspendOnly, boolean forced) {
+ if (forced &&
"stop-context".equals(Thread.currentThread().getName())
+ && holdForcedPrepare.compareAndSet(true, false)) {
+ inForcedPrepare.countDown();
+ try {
+ assertTrue(releaseForcedPrepare.await(30,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ }
+
+ private static final class SuspendableConsumer extends DefaultConsumer
implements Suspendable {
+ SuspendableConsumer(Endpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ }
+ }
+}
