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 ba87a347b68f CAMEL-26104: camel-core - ThrottlingExceptionRoutePolicy:
create the half-open timer through the ExecutorServiceManager (#28753)
ba87a347b68f is described below
commit ba87a347b68fbd98728b21802f04c73430c1510d
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 31 15:32:18 2026 +0530
CAMEL-26104: camel-core - ThrottlingExceptionRoutePolicy: create the
half-open timer through the ExecutorServiceManager (#28753)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/test-dst.adoc | 10 +-
.../camel-test-dst/src/main/docs/test-dst.adoc | 10 +-
.../camel/test/dst/DstHalfOpenCheckTest.java | 126 +++++++++
...tlingExceptionRoutePolicyFailureWindowTest.java | 36 ++-
...gExceptionRoutePolicyHalfOpenSchedulerTest.java | 294 +++++++++++++++++++++
.../throttling/ThrottlingExceptionRoutePolicy.java | 165 ++++++++++--
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 16 ++
7 files changed, 618 insertions(+), 39 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
index 9c0d809a57ad..2e7e4906b15f 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
@@ -92,9 +92,10 @@ the synchronous redelivery (`RedeliveryPolicy`,
`RedeliveryErrorHandler`), `Fore
the budgets of both tasks, the wait loop of `DefaultShutdownStrategy`, the
synchronous delayer, `MulticastProcessor`,
`OptimisticLockRetryPolicy`, `DefaultStartupConditionStrategy`, `StopWatch`,
`MonotonicClock`, `BackOffTimerTask`,
`DefaultTimeoutMap`, the clocks of the throttlers and the counters of
`ThrottlingExceptionRoutePolicy`. A sleep there is
-a virtual wait on the loop thread, traced with its call site. Three waits of
these classes are not virtual (see below):
-the permit queue of `TotalRequestsThrottler`, the `Semaphore` of
`ConcurrentRequestsThrottler` and the half-open timer
-of `ThrottlingExceptionRoutePolicy`. The sleep of the route `delayer` option
is not virtual either (see below).
+a virtual wait on the loop thread, traced with its call site. Two waits of
these classes are not virtual (see below):
+the permit queue of `TotalRequestsThrottler` and the `Semaphore` of
`ConcurrentRequestsThrottler`. The sleep of the
+route `delayer` option is not virtual either (see below). The half-open check
of `ThrottlingExceptionRoutePolicy` runs
+on a scheduler of the `ExecutorServiceManager`, so the circuit is half opened
after `halfOpenAfter` of virtual time.
|===
Event timestamps, the clock of an exchange and the direct clock reads of the
debugger, the tracer, management and the
@@ -137,9 +138,6 @@ core:
* The route `delayer` option (`CamelInternalProcessor.DelayerAdvice`) calls
`Thread.sleep` directly, not through
`TimeSource`: the delay takes real time on the loop thread and the virtual
clock does not move. Use the `delay` EIP
in a world instead.
-* The half-open check of `ThrottlingExceptionRoutePolicy` runs on a
`java.util.Timer` thread, outside the harness. The
- timer fires after `halfOpenAfter` of real time (30 s by default, longer than
the watchdog) and its clock read is an
- escape, so a world does not see the circuit half-open in virtual time.
=== The one-stack model
diff --git a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
index 9c0d809a57ad..2e7e4906b15f 100644
--- a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
+++ b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
@@ -92,9 +92,10 @@ the synchronous redelivery (`RedeliveryPolicy`,
`RedeliveryErrorHandler`), `Fore
the budgets of both tasks, the wait loop of `DefaultShutdownStrategy`, the
synchronous delayer, `MulticastProcessor`,
`OptimisticLockRetryPolicy`, `DefaultStartupConditionStrategy`, `StopWatch`,
`MonotonicClock`, `BackOffTimerTask`,
`DefaultTimeoutMap`, the clocks of the throttlers and the counters of
`ThrottlingExceptionRoutePolicy`. A sleep there is
-a virtual wait on the loop thread, traced with its call site. Three waits of
these classes are not virtual (see below):
-the permit queue of `TotalRequestsThrottler`, the `Semaphore` of
`ConcurrentRequestsThrottler` and the half-open timer
-of `ThrottlingExceptionRoutePolicy`. The sleep of the route `delayer` option
is not virtual either (see below).
+a virtual wait on the loop thread, traced with its call site. Two waits of
these classes are not virtual (see below):
+the permit queue of `TotalRequestsThrottler` and the `Semaphore` of
`ConcurrentRequestsThrottler`. The sleep of the
+route `delayer` option is not virtual either (see below). The half-open check
of `ThrottlingExceptionRoutePolicy` runs
+on a scheduler of the `ExecutorServiceManager`, so the circuit is half opened
after `halfOpenAfter` of virtual time.
|===
Event timestamps, the clock of an exchange and the direct clock reads of the
debugger, the tracer, management and the
@@ -137,9 +138,6 @@ core:
* The route `delayer` option (`CamelInternalProcessor.DelayerAdvice`) calls
`Thread.sleep` directly, not through
`TimeSource`: the delay takes real time on the loop thread and the virtual
clock does not move. Use the `delay` EIP
in a world instead.
-* The half-open check of `ThrottlingExceptionRoutePolicy` runs on a
`java.util.Timer` thread, outside the harness. The
- timer fires after `halfOpenAfter` of real time (30 s by default, longer than
the watchdog) and its clock read is an
- escape, so a world does not see the circuit half-open in virtual time.
=== The one-stack model
diff --git
a/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstHalfOpenCheckTest.java
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstHalfOpenCheckTest.java
new file mode 100644
index 000000000000..3828b9f4a5d2
--- /dev/null
+++
b/components/camel-test/camel-test-dst/src/test/java/org/apache/camel/test/dst/DstHalfOpenCheckTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.test.dst;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
+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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The half open check of {@code ThrottlingExceptionRoutePolicy} runs on a
scheduler created through the
+ * ExecutorServiceManager: in a run it is a task of the loop, and the circuit
is half opened after {@code halfOpenAfter}
+ * of virtual time. When {@code halfOpenAfter} is raised while the circuit is
open, the check schedules the next one for
+ * the time that is left.
+ */
+@DstInstrument("org.apache.camel.test.dst.sample")
+class DstHalfOpenCheckTest {
+
+ private static final long REAL_LIMIT = 5_000_000_000L;
+
+ @Test
+ void theCircuitIsHalfOpenedAfterHalfOpenAfterOfVirtualTime() throws
Exception {
+ Dst dst = new Dst(1);
+ dst.strict(true, List.of());
+ DefaultCamelContext context = DstCamel.newContext(dst);
+ try {
+ ThrottlingExceptionRoutePolicy policy = new
ThrottlingExceptionRoutePolicy(1, 60_000, 30_000, List.of());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:in").routeId("r").routePolicy(policy)
+ .throwException(new IllegalStateException("the
downstream system is down"));
+ }
+ });
+ context.start();
+ Exchange[] done = new Exchange[1];
+ dst.executor("client", 1).schedule("send", 1_000, () ->
send(context, done));
+ long start = System.nanoTime();
+
+ dst.runFor(30_000);
+ assertNotNull(done[0]);
+ assertEquals("opened", policy.getStateAsString());
+ assertTrue(((ServiceSupport)
context.getRoute("r").getConsumer()).isSuspended());
+
+ dst.runUntil(() -> "half
opened".equals(policy.getStateAsString()), "the circuit is half opened");
+ assertEquals(31_000, dst.now(), "half opened 30 s after the
failure at 1 s");
+ assertFalse(((ServiceSupport)
context.getRoute("r").getConsumer()).isSuspended());
+ assertTrue(dst.traceLines().stream()
+ .anyMatch(l -> l.contains("thread pool created:
ThrottlingExceptionRoutePolicy (1 threads)")),
+ String.join("\n", dst.traceLines()));
+ assertTrue(System.nanoTime() - start < REAL_LIMIT, "30 s of
virtual time take no real time");
+ } finally {
+ dst.dropPending();
+ context.stop();
+ dst.close();
+ }
+ }
+
+ @Test
+ void aRaisedHalfOpenAfterHalfOpensTheCircuitWhenTheNewTimeIsUp() throws
Exception {
+ Dst dst = new Dst(1);
+ dst.strict(true, List.of());
+ DefaultCamelContext context = DstCamel.newContext(dst);
+ try {
+ ThrottlingExceptionRoutePolicy policy = new
ThrottlingExceptionRoutePolicy(1, 60_000, 30_000, List.of());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:in").routeId("r").routePolicy(policy)
+ .throwException(new IllegalStateException("the
downstream system is down"));
+ }
+ });
+ context.start();
+ Exchange[] done = new Exchange[1];
+ DstExecutor client = dst.executor("client", 1);
+ client.schedule("send", 1_000, () -> send(context, done));
+ // as over JMX, while the circuit is open: the check scheduled at
1 s is due at 31 s, before the new time
+ client.schedule("raise halfOpenAfter", 20_000, () ->
policy.setHalfOpenAfter(40_000));
+
+ dst.runFor(31_000);
+ assertEquals("opened", policy.getStateAsString());
+ assertTrue(((ServiceSupport)
context.getRoute("r").getConsumer()).isSuspended());
+
+ // the check at 31 s schedules the next one for the 10 s that are
left
+ dst.runUntil(() -> "half
opened".equals(policy.getStateAsString()), "the circuit is half opened");
+ assertEquals(41_000, dst.now(), "half opened 40 s after the
failure at 1 s");
+ assertFalse(((ServiceSupport)
context.getRoute("r").getConsumer()).isSuspended());
+ } finally {
+ dst.dropPending();
+ context.stop();
+ dst.close();
+ }
+ }
+
+ /** Sends an exchange to the route's consumer without waiting for it (no
blocking wait on the loop thread). */
+ private static void send(DefaultCamelContext context, Exchange[] done) {
+ DefaultConsumer consumer = (DefaultConsumer)
context.getRoute("r").getConsumer();
+ Exchange exchange = consumer.createExchange(false);
+ exchange.getMessage().setBody("hello");
+ consumer.getAsyncProcessor().process(exchange, doneSync -> done[0] =
exchange);
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyFailureWindowTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyFailureWindowTest.java
index d14b861661f5..72ab905d2c21 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyFailureWindowTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyFailureWindowTest.java
@@ -19,12 +19,17 @@ package org.apache.camel.processor.throttle;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import org.apache.camel.CamelContext;
import org.apache.camel.CamelExecutionException;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Route;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.ThreadPoolProfile;
+import org.apache.camel.support.DefaultThreadPoolFactory;
import org.apache.camel.support.service.ServiceSupport;
import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
import org.apache.camel.util.TimeSource;
@@ -56,6 +61,25 @@ class ThrottlingExceptionRoutePolicyFailureWindowTest
extends ContextTestSupport
private final CountDownLatch slowEntered = new CountDownLatch(1);
private final CountDownLatch slowRelease = new CountDownLatch(1);
+ // the scheduler of the half open checks of the policy of the race route
(the only policy whose circuit opens there)
+ private volatile ScheduledExecutorService raceScheduler;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getExecutorServiceManager().setThreadPoolFactory(new
DefaultThreadPoolFactory() {
+ @Override
+ public ScheduledExecutorService
newScheduledThreadPool(ThreadPoolProfile profile, ThreadFactory threadFactory) {
+ ScheduledExecutorService answer =
super.newScheduledThreadPool(profile, threadFactory);
+ if ("ThrottlingExceptionRoutePolicy".equals(profile.getId())) {
+ raceScheduler = answer;
+ }
+ return answer;
+ }
+ });
+ return context;
+ }
+
@Test
void testSuccessInHalfOpenStateClosesTheCircuit() throws Exception {
ServiceSupport consumer = (ServiceSupport)
context.getRoute("halfOpen").getConsumer();
@@ -120,10 +144,10 @@ class ThrottlingExceptionRoutePolicyFailureWindowTest
extends ContextTestSupport
assertEquals("half opened", racePolicy.getStateAsString());
assertFalse(consumer.isSuspended());
- // the timer's check goes on and ends
+ // the timer's check goes on and ends: the scheduler has one thread,
so a task submitted now runs after it
racePolicy.timerMayGo.countDown();
- racePolicy.timerThread.join(10000);
- assertFalse(racePolicy.timerThread.isAlive());
+ raceScheduler.submit(() -> {
+ }).get(10, TimeUnit.SECONDS);
assertEquals("half opened", racePolicy.getStateAsString(),
"The half open timer should not open the circuit again after
an exchange half opened it");
assertFalse(consumer.isSuspended());
@@ -158,8 +182,8 @@ class ThrottlingExceptionRoutePolicyFailureWindowTest
extends ContextTestSupport
}
/**
- * The policy, with latches: halfOpened when the circuit has been half
opened, and a gate that stops the half open
- * check of the timer thread (the first one) before it reads the state,
until timerMayGo.
+ * The policy, with latches: halfOpened when the circuit has been half
opened, and a gate that stops the first half
+ * open check (on the thread of the policy's scheduler) before it reads
the state, until timerMayGo.
*/
private static final class LatchPolicy extends
ThrottlingExceptionRoutePolicy {
private final CountDownLatch halfOpened = new CountDownLatch(1);
@@ -174,7 +198,7 @@ class ThrottlingExceptionRoutePolicyFailureWindowTest
extends ContextTestSupport
@Override
protected boolean isThresholdExceeded() {
// the half open check of the timer calls this first, before it
reads the state
- if (timerThread == null &&
Thread.currentThread().getName().startsWith("Timer-")) {
+ if (timerThread == null &&
Thread.currentThread().getName().endsWith(" - ThrottlingExceptionRoutePolicy"))
{
timerThread = Thread.currentThread();
timerChecking.countDown();
try {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenSchedulerTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenSchedulerTest.java
new file mode 100644
index 000000000000..a92961c40d0f
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenSchedulerTest.java
@@ -0,0 +1,294 @@
+/*
+ * 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.processor.throttle;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.log.ConsumingAppender;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.Appender;
+import org.apache.logging.log4j.core.LoggerContext;
+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.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The half open check of {@link ThrottlingExceptionRoutePolicy} runs on one
thread of a scheduler that the policy
+ * creates through the ExecutorServiceManager, and that thread ends when the
last route of the policy is removed. An
+ * exception of the check, or of suspending or resuming the consumer, goes to
the exception handler of the policy.
+ */
+class ThrottlingExceptionRoutePolicyHalfOpenSchedulerTest extends
ContextTestSupport {
+
+ private final LatchPolicy policy = new LatchPolicy(100, 2);
+ private final LatchPolicy removePolicy = new LatchPolicy(60000, 1);
+ private final LatchPolicy sharedPolicy = new LatchPolicy(3000, 1);
+ private final FailingSuspendPolicy suspendPolicy = new
FailingSuspendPolicy();
+ private final LatchPolicy handlerPolicy = new LatchPolicy(100, 1);
+ private final List<Throwable> handled = new CopyOnWriteArrayList<>();
+ private final AtomicInteger handlerChecks = new AtomicInteger();
+
+ @Test
+ void testHalfOpenChecksRunOnOneThreadOfThePolicy() throws Exception {
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("checked").getConsumer();
+
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:checked", "fail"));
+ assertEquals("opened", policy.getStateAsString());
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
policy.checkThreads.size() == 1);
+ assertEquals("half opened", policy.getStateAsString());
+ assertFalse(consumer.isSuspended());
+
+ // a failure in the half open state opens the circuit again, and the
next check runs on the same thread
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:checked", "fail"));
+ assertTrue(policy.halfOpened.await(10, TimeUnit.SECONDS));
+ assertEquals("half opened", policy.getStateAsString());
+
+ Thread thread = policy.checkThreads.get(0);
+ assertSame(thread, policy.checkThreads.get(1));
+ assertTrue(isSchedulerThread(thread), thread.getName());
+ }
+
+ @Test
+ void testRemovingTheRouteEndsTheThreadOfTheHalfOpenCheck() throws
Exception {
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:removed", "fail"));
+ assertEquals("opened", removePolicy.getStateAsString());
+ assertEquals(1, schedulerThreads().size(), "the half open check is
scheduled");
+
+ context.getRouteController().stopRoute("removed");
+ assertTrue(context.removeRoute("removed"));
+
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
schedulerThreads().isEmpty());
+ assertEquals(1, removePolicy.halfOpened.getCount(), "the check
scheduled before the removal must not run");
+ }
+
+ @Test
+ void testRemovingOneOfTwoRoutesOfAPolicyKeepsItsHalfOpenCheck() throws
Exception {
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("shared2").getConsumer();
+
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:shared2", "fail"));
+ assertEquals("opened", sharedPolicy.getStateAsString());
+ assertTrue(consumer.isSuspended());
+
+ context.getRouteController().stopRoute("shared1");
+ assertTrue(context.removeRoute("shared1"));
+ assertEquals(1, sharedPolicy.halfOpened.getCount(), "the check is
still pending when the other route is removed");
+
+ // the route that is left still gets its half open check
+ assertTrue(sharedPolicy.halfOpened.await(10, TimeUnit.SECONDS));
+ assertFalse(consumer.isSuspended());
+
+ context.getRouteController().stopRoute("shared2");
+ assertTrue(context.removeRoute("shared2"));
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
schedulerThreads().isEmpty());
+ }
+
+ @Test
+ void testRestartedContextKeepsTheHalfOpenCheckOfTheRouteThatIsLeft()
throws Exception {
+ // a CamelContext that is stopped calls onRemove of the policy for
each route, and when it is started again it
+ // creates the routes again (onInit and onStart)
+ context.stop();
+ context.start();
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("shared2").getConsumer();
+
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:shared2", "fail"));
+ assertEquals("opened", sharedPolicy.getStateAsString());
+ assertTrue(consumer.isSuspended());
+
+ context.getRouteController().stopRoute("shared1");
+ assertTrue(context.removeRoute("shared1"));
+ assertEquals(1, sharedPolicy.halfOpened.getCount(), "the check is
still pending when the other route is removed");
+
+ // the route that is left still gets its half open check
+ assertTrue(sharedPolicy.halfOpened.await(10, TimeUnit.SECONDS));
+ assertFalse(consumer.isSuspended());
+ }
+
+ @Test
+ void testFailureToSuspendTheConsumerIsLogged() throws Exception {
+ // the policy logs with the LoggingExceptionHandler that
RoutePolicySupport.onInit creates
+ String loggerName = FailingSuspendPolicy.class.getName();
+ List<Throwable> logged = new CopyOnWriteArrayList<>();
+ Appender appender = ConsumingAppender.newAppender(loggerName,
"suspend-failure", Level.WARN, e -> {
+ if (e.getThrown() != null) {
+ logged.add(e.getThrown());
+ }
+ });
+ try {
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:suspendFails", "fail"));
+
+ assertEquals(1, logged.size(), "the failure to suspend the
consumer is logged");
+ assertEquals("Forced suspend failure", logged.get(0).getMessage());
+ } finally {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ ctx.getConfiguration().removeLogger(loggerName);
+ ctx.updateLoggers();
+ appender.stop();
+ }
+ }
+
+ @Test
+ void testExceptionOfTheHalfOpenHandlerIsHandledAndTheCheckIsRetried()
throws Exception {
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("handler").getConsumer();
+
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:handler", "fail"));
+ assertEquals("opened", handlerPolicy.getStateAsString());
+
+ // the first check fails, and keeps the circuit open; the next one,
after halfOpenAfter, closes it
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
"closed".equals(handlerPolicy.getStateAsString()));
+ assertEquals(2, handlerChecks.get());
+ assertEquals(1, handled.size(), "the exception of the half open
handler is handled");
+ assertEquals("Forced check failure", handled.get(0).getMessage());
+ assertFalse(consumer.isSuspended());
+ }
+
+ @Test
+ void testHalfOpenAfterMustBeGreaterThanZero() throws Exception {
+ ThrottlingExceptionRoutePolicy configured = new
ThrottlingExceptionRoutePolicy();
+ assertThrows(IllegalArgumentException.class, () ->
configured.setHalfOpenAfter(0));
+ assertThrows(IllegalArgumentException.class, () ->
configured.setHalfOpenAfter(-1));
+ assertEquals(30000, configured.getHalfOpenAfter());
+
+ // set through the constructor: the route that uses the policy is not
created
+ ThrottlingExceptionRoutePolicy constructed = new
ThrottlingExceptionRoutePolicy(1, 60000, 0, List.of());
+ Exception e = assertThrows(Exception.class, () ->
context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:zero").routeId("zero").routePolicy(constructed).to("mock:zero");
+ }
+ }));
+ Throwable cause = e;
+ while (cause != null && !(cause instanceof IllegalArgumentException)) {
+ cause = cause.getCause();
+ }
+ assertEquals("halfOpenAfter must be greater than 0, was: 0", cause !=
null ? cause.getMessage() : null, e.toString());
+ }
+
+ private boolean isSchedulerThread(Thread thread) {
+ // the thread name pattern of the ExecutorServiceManager, with the
name of this CamelContext
+ String name = thread.getName();
+ return name.startsWith("Camel (" + context.getName() + ") thread #")
+ && name.endsWith(" - ThrottlingExceptionRoutePolicy");
+ }
+
+ private List<Thread> schedulerThreads() {
+ return
Thread.getAllStackTraces().keySet().stream().filter(Thread::isAlive).filter(this::isSchedulerThread)
+ .toList();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:checked").routeId("checked").routePolicy(policy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+
+
from("direct:removed").routeId("removed").routePolicy(removePolicy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+
+
from("direct:shared1").routeId("shared1").routePolicy(sharedPolicy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+
from("direct:shared2").routeId("shared2").routePolicy(sharedPolicy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+
+
from("direct:suspendFails").routeId("suspendFails").routePolicy(suspendPolicy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+
+ handlerPolicy.setExceptionHandler(new
RecordingExceptionHandler());
+ handlerPolicy.setHalfOpenHandler(() -> {
+ if (handlerChecks.incrementAndGet() == 1) {
+ throw new IllegalStateException("Forced check
failure");
+ }
+ return true;
+ });
+
from("direct:handler").routeId("handler").routePolicy(handlerPolicy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+ }
+ };
+ }
+
+ /**
+ * The policy, with the threads of its half open checks and a latch
counted down when the circuit is half opened.
+ */
+ private static final class LatchPolicy extends
ThrottlingExceptionRoutePolicy {
+ private final List<Thread> checkThreads = new CopyOnWriteArrayList<>();
+ private final CountDownLatch halfOpened;
+
+ LatchPolicy(long halfOpenAfter, int halfOpenings) {
+ super(1, 60000, halfOpenAfter, List.of());
+ this.halfOpened = new CountDownLatch(halfOpenings);
+ }
+
+ @Override
+ protected void halfOpenCircuit(Route route) {
+ super.halfOpenCircuit(route);
+ checkThreads.add(Thread.currentThread());
+ halfOpened.countDown();
+ }
+ }
+
+ /**
+ * The policy, failing to suspend the consumer.
+ */
+ private static final class FailingSuspendPolicy extends
ThrottlingExceptionRoutePolicy {
+
+ FailingSuspendPolicy() {
+ super(1, 60000, 60000, List.of());
+ }
+
+ @Override
+ public boolean suspendOrStopConsumer(Consumer consumer) {
+ throw new IllegalStateException("Forced suspend failure");
+ }
+ }
+
+ private final class RecordingExceptionHandler implements ExceptionHandler {
+
+ @Override
+ public void handleException(Throwable exception) {
+ handled.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handled.add(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ handled.add(exception);
+ }
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
index 255bcd7be538..9ad737f8da7a 100644
---
a/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
+++
b/core/camel-support/src/main/java/org/apache/camel/throttling/ThrottlingExceptionRoutePolicy.java
@@ -19,9 +19,12 @@ package org.apache.camel.throttling;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
+import java.util.HashSet;
import java.util.List;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
@@ -102,7 +105,16 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
private final AtomicInteger success = new AtomicInteger();
private final AtomicInteger state = new AtomicInteger(STATE_CLOSED);
private final AtomicBoolean keepOpenBool = new AtomicBoolean();
- private volatile Timer halfOpenTimer;
+ // the half open check: one at a time, on a scheduler created through the
ExecutorServiceManager when the circuit
+ // opens the first time, and shut down when the last route of the policy
is removed or the policy is stopped;
+ // guarded by schedulerLock (not by lock, which a running check holds
while it suspends or resumes the consumer)
+ private final Lock schedulerLock = new ReentrantLock();
+ private final Set<String> routeIds = new HashSet<>();
+ private volatile ScheduledExecutorService halfOpenScheduler;
+ private volatile ScheduledFuture<?> halfOpenCheck;
+ // set when the scheduler is shut down, so that a check that is running
then does not create a new one; cleared
+ // when the policy or a route of it is started again
+ private boolean halfOpenStopped;
private volatile long lastFailure;
private volatile long openedAt;
// the times of the most recent failures (at most failureThreshold of
them), oldest first, guarded by failureLock
@@ -165,6 +177,10 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
this.stateLogger = new CamelLogger(LOG, stateLoggingLevel);
+ // with 0 or less, a check that keeps the circuit open (keepOpen, or a
halfOpenHandler that is not ready)
+ // would schedule the next check without delay, a busy loop on the
scheduler thread
+ checkHalfOpenAfter(halfOpenAfter);
+
if (exceptions != null && throttledExceptions == null) {
var list = new ArrayList<Class<?>>();
for (String fqn : exceptions.split(",")) {
@@ -178,11 +194,20 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
@Override
public void onInit(Route route) {
LOG.debug("Initializing ThrottlingExceptionRoutePolicy route policy");
+ // creates the exception handler that logs a failure to suspend or
resume the consumer
+ super.onInit(route);
+ if (camelContext == null) {
+ camelContext = route.getCamelContext();
+ }
+ addRoute(route);
logState();
}
@Override
public void onStart(Route route) {
+ // also here, not only in onInit, so that a started route is always
tracked (a CamelContext that is stopped
+ // calls onRemove for each route; when started again it creates the
routes again and calls onInit too)
+ addRoute(route);
// if keepOpen then start w/ the circuit open
if (keepOpenBool.get()) {
// the circuit may still be open from before the route was
stopped, and then the started consumer must
@@ -191,12 +216,66 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
}
+ @Override
+ public void onRemove(Route route) {
+ // the policy is a service of the CamelContext, which does not stop it
when its routes are removed
+ schedulerLock.lock();
+ try {
+ routeIds.remove(route.getRouteId());
+ if (routeIds.isEmpty()) {
+ shutdownHalfOpenScheduler();
+ }
+ } finally {
+ schedulerLock.unlock();
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ allowHalfOpenChecks();
+ }
+
@Override
protected void doStop() throws Exception {
- Timer timer = halfOpenTimer;
- if (timer != null) {
- timer.cancel();
- halfOpenTimer = null;
+ shutdownHalfOpenScheduler();
+ }
+
+ private void addRoute(Route route) {
+ schedulerLock.lock();
+ try {
+ routeIds.add(route.getRouteId());
+ halfOpenStopped = false;
+ } finally {
+ schedulerLock.unlock();
+ }
+ }
+
+ private void allowHalfOpenChecks() {
+ schedulerLock.lock();
+ try {
+ halfOpenStopped = false;
+ } finally {
+ schedulerLock.unlock();
+ }
+ }
+
+ private void shutdownHalfOpenScheduler() {
+ schedulerLock.lock();
+ try {
+ halfOpenStopped = true;
+ ScheduledFuture<?> check = halfOpenCheck;
+ if (check != null) {
+ check.cancel(false);
+ halfOpenCheck = null;
+ }
+ ScheduledExecutorService scheduler = halfOpenScheduler;
+ if (scheduler != null) {
+ halfOpenScheduler = null;
+ // does not wait for a check that is running
+ camelContext.getExecutorServiceManager().shutdown(scheduler);
+ }
+ } finally {
+ schedulerLock.unlock();
}
}
@@ -327,7 +406,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
if (openFor <= elapsedTimeSinceOpened) {
LOG.debug("Checking an open circuit...");
if (halfOpenHandler != null) {
- if (halfOpenHandler.isReadyToBeClosed()) {
+ if (isReadyToBeClosed()) {
LOG.debug("Closing circuit...");
closeCircuit(route);
} else {
@@ -356,6 +435,16 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
+ private boolean isReadyToBeClosed() {
+ try {
+ return halfOpenHandler.isReadyToBeClosed();
+ } catch (Exception e) {
+ // the circuit is kept open, and checked again after halfOpenAfter
+ handleException(e);
+ return false;
+ }
+ }
+
protected boolean isThresholdExceeded() {
boolean output = false;
logState();
@@ -414,6 +503,10 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
}
+ /**
+ * Schedules the check of the open circuit after {@link
#getHalfOpenAfter()} milliseconds, and cancels the check
+ * scheduled before, if any.
+ */
protected void addHalfOpenTimer(Route route) {
addHalfOpenTimer(route, halfOpenAfter);
}
@@ -423,16 +516,27 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
*/
protected void addHalfOpenTimer(Route route, long delay) {
lock.lock();
+ schedulerLock.lock();
try {
- Timer previous = halfOpenTimer;
- Timer timer = new Timer();
- timer.schedule(new HalfOpenTask(route, timer), delay);
- halfOpenTimer = timer;
+ if (halfOpenStopped) {
+ LOG.debug("Not scheduling a half open check (the route is
removed or the policy is stopped)");
+ return;
+ }
+ ScheduledExecutorService scheduler = halfOpenScheduler;
+ if (scheduler == null || scheduler.isShutdown()) {
+ // one thread for the half open checks of this policy, named
and managed by Camel
+ scheduler =
camelContext.getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "ThrottlingExceptionRoutePolicy");
+ halfOpenScheduler = scheduler;
+ }
+ ScheduledFuture<?> previous = halfOpenCheck;
+ halfOpenCheck = scheduler.schedule(new HalfOpenTask(route,
scheduler), delay, TimeUnit.MILLISECONDS);
if (previous != null) {
- // only one half open check at a time (each timer has its own
thread)
- previous.cancel();
+ // only one half open check at a time (a check that is running
is not interrupted)
+ previous.cancel(false);
}
} finally {
+ schedulerLock.unlock();
lock.unlock();
}
}
@@ -506,20 +610,27 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
}
- class HalfOpenTask extends TimerTask {
+ class HalfOpenTask implements Runnable {
private final Route route;
- private final Timer timer;
+ private final ScheduledExecutorService scheduler;
- HalfOpenTask(Route route, Timer timer) {
+ HalfOpenTask(Route route, ScheduledExecutorService scheduler) {
this.route = route;
- this.timer = timer;
+ this.scheduler = scheduler;
}
@Override
public void run() {
- // cancel the timer of this task only, a newer timer may already
have been scheduled
- timer.cancel();
- calculateState(route, null);
+ if (scheduler != halfOpenScheduler) {
+ // the route was removed or the policy stopped since this
check was scheduled
+ return;
+ }
+ try {
+ calculateState(route, null);
+ } catch (Exception e) {
+ // the scheduler would keep it in the future of this check,
where no one sees it
+ handleException(e);
+ }
}
}
@@ -578,10 +689,22 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
return halfOpenAfter;
}
+ /**
+ * Sets the interval (in millis) for how often to check whether a
currently open circuit breaker may work again.
+ *
+ * @throws IllegalArgumentException if the interval is 0 or less
+ */
public void setHalfOpenAfter(long halfOpenAfter) {
+ checkHalfOpenAfter(halfOpenAfter);
this.halfOpenAfter = halfOpenAfter;
}
+ private static void checkHalfOpenAfter(long halfOpenAfter) {
+ if (halfOpenAfter <= 0) {
+ throw new IllegalArgumentException("halfOpenAfter must be greater
than 0, was: " + halfOpenAfter);
+ }
+ }
+
public int getFailures() {
return failures.get();
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index ee38141b1a57..0ff788fe9f49 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -247,6 +247,22 @@ between `next()` and `remove()` must call `remove()`
before `hasNext()`.
outside the Basic Multilingual Plane (such as U+10400 and U+10428) as the same
key, as `String.equalsIgnoreCase` does;
before, they were usually kept as two entries. Keys without such letters are
not affected.
+=== camel-core - ThrottlingExceptionRoutePolicy runs the half open check on a
thread pool of Camel
+
+The half open check of `ThrottlingExceptionRoutePolicy` now runs on a thread
pool created through the
+`ExecutorServiceManager` (one thread, named `ThrottlingExceptionRoutePolicy`
and managed like the other thread pools of
+Camel), instead of a new `java.util.Timer` thread each time the circuit opens.
The thread is created when the circuit
+opens the first time, and it ends when the last route of the policy is removed
or the policy is stopped.
+
+`halfOpenAfter` must now be greater than 0: adding a route with a policy whose
`halfOpenAfter` is 0 or less fails,
+and `setHalfOpenAfter` (also over JMX) rejects such a value with an
`IllegalArgumentException`. With 0, a check that
+keeps the circuit open (`keepOpen`, or a `halfOpenHandler` that is not ready)
scheduled the next check without delay, a
+busy loop. To half open the circuit almost at once, as 0 did, use a small
value such as 1.
+
+An exception thrown by the `isReadyToBeClosed` method of a `halfOpenHandler`
is now logged (by the exception handler of
+the policy), and the circuit is kept open and checked again after
`halfOpenAfter`; before, the timer thread died and the
+circuit stayed open.
+
=== camel-main - a failure to start is logged
A Camel Main application that fails to start now logs why, at ERROR: `Error
starting Camel: ...` with the stack trace,
