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 295b8d648cf2 CAMEL-26291: camel-core - ThrottlingExceptionRoutePolicy
must count the failures within failureWindow, and close the circuit on a
success in the half open state (#28467)
295b8d648cf2 is described below
commit 295b8d648cf2014ee671f264634a31e60a027386
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 26 13:32:36 2026 +0530
CAMEL-26291: camel-core - ThrottlingExceptionRoutePolicy must count the
failures within failureWindow, and close the circuit on a success in the half
open state (#28467)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
...tlingExceptionRoutePolicyFailureWindowTest.java | 195 +++++++++++++++++++++
...xceptionRoutePolicyHalfOpenHandlerSedaTest.java | 2 +-
...ingExceptionRoutePolicyHalfOpenHandlerTest.java | 2 +-
...ThrottlingExceptionRoutePolicyHalfOpenTest.java | 2 +-
.../ThrottlingExceptionRoutePolicyTest.java | 2 +-
.../throttling/ThrottlingExceptionRoutePolicy.java | 88 ++++++++--
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++
7 files changed, 286 insertions(+), 18 deletions(-)
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
new file mode 100644
index 000000000000..d14b861661f5
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyFailureWindowTest.java
@@ -0,0 +1,195 @@
+/*
+ * 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.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+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.support.service.ServiceSupport;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
+import org.apache.camel.util.TimeSource;
+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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The {@link ThrottlingExceptionRoutePolicy} opens the circuit when
failureThreshold failures occur within the
+ * failureWindow, and an exchange done in the half open state decides: a
success closes the circuit, also when the
+ * failures that opened it are still within the failure window.
+ */
+class ThrottlingExceptionRoutePolicyFailureWindowTest extends
ContextTestSupport {
+
+ // long enough for two sends in a row to be within it also on a slow
machine
+ private static final long FAILURE_WINDOW = 1000;
+
+ // the half open check comes before the failures that opened the circuit
leave the window (as with the defaults: a
+ // window of 60 seconds and a half open check after 30 seconds)
+ private final LatchPolicy halfOpenPolicy = new LatchPolicy(1, 60000, 100);
+ private final ThrottlingExceptionRoutePolicy windowPolicy
+ = new ThrottlingExceptionRoutePolicy(2, FAILURE_WINDOW, 60000,
null);
+ private final LatchPolicy racePolicy = new LatchPolicy(1, 60000, 100);
+
+ private final CountDownLatch slowEntered = new CountDownLatch(1);
+ private final CountDownLatch slowRelease = new CountDownLatch(1);
+
+ @Test
+ void testSuccessInHalfOpenStateClosesTheCircuit() throws Exception {
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("halfOpen").getConsumer();
+
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:halfOpen", "fail"));
+ // the half open check of the timer waits until it may go, so the
circuit is still open
+ assertEquals("opened", halfOpenPolicy.getStateAsString());
+ assertTrue(consumer.isSuspended());
+
+ // the half open timer resumes the consumer
+ halfOpenPolicy.timerMayGo.countDown();
+ assertTrue(halfOpenPolicy.halfOpened.await(10, TimeUnit.SECONDS));
+ assertEquals("half opened", halfOpenPolicy.getStateAsString());
+ // no further half open check while the test runs
+ halfOpenPolicy.setHalfOpenAfter(60000);
+
+ template.sendBody("direct:halfOpen", "ok");
+ assertEquals("closed", halfOpenPolicy.getStateAsString(), "A success
in the half open state should close the circuit");
+ assertFalse(consumer.isSuspended());
+ }
+
+ @Test
+ void testFailuresOutsideTheWindowDoNotOpenTheCircuit() {
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("window").getConsumer();
+
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:window", "fail"));
+ long first = windowPolicy.getLastFailure();
+ // the first failure leaves the window (a condition on the clock,
there is no event to wait for)
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
TimeSource.currentTimeMillis() > first + FAILURE_WINDOW);
+
+ // only this failure is within the window
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:window", "fail"));
+ assertEquals("closed", windowPolicy.getStateAsString(),
+ "One failure within the window should not open the circuit
with a threshold of 2");
+ assertFalse(consumer.isSuspended());
+
+ // and a second failure within the window opens it
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:window", "fail"));
+ assertEquals("opened", windowPolicy.getStateAsString());
+ assertTrue(consumer.isSuspended());
+ }
+
+ /**
+ * An exchange that was in flight when the circuit opened is done
successfully when the half open check of the timer
+ * is due, and half opens the circuit itself; the timer's check, which
runs at the same time, must then leave the
+ * half open circuit to the next exchange, and not open it again because
of the failure that opened it.
+ */
+ @Test
+ void testHalfOpenTimerDoesNotReopenACircuitHalfOpenedByAnExchange() throws
Exception {
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("race").getConsumer();
+
+ Future<Object> slow = template.asyncRequestBody("direct:race", "slow");
+ assertTrue(slowEntered.await(10, TimeUnit.SECONDS));
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:race", "fail"));
+ assertEquals("opened", racePolicy.getStateAsString());
+ assertTrue(consumer.isSuspended());
+
+ // the timer's half open check has started (and waits); the straggler
is done with a success now
+ assertTrue(racePolicy.timerChecking.await(10, TimeUnit.SECONDS));
+ slowRelease.countDown();
+ assertEquals("slow", slow.get(10, TimeUnit.SECONDS));
+ assertEquals("half opened", racePolicy.getStateAsString());
+ assertFalse(consumer.isSuspended());
+
+ // the timer's check goes on and ends
+ racePolicy.timerMayGo.countDown();
+ racePolicy.timerThread.join(10000);
+ assertFalse(racePolicy.timerThread.isAlive());
+ assertEquals("half opened", racePolicy.getStateAsString(),
+ "The half open timer should not open the circuit again after
an exchange half opened it");
+ assertFalse(consumer.isSuspended());
+
+ // the next exchange decides
+ template.sendBody("direct:race", "ok");
+ assertEquals("closed", racePolicy.getStateAsString());
+ assertFalse(consumer.isSuspended());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:halfOpen").routeId("halfOpen").routePolicy(halfOpenPolicy)
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+
+
from("direct:window").routeId("window").routePolicy(windowPolicy)
+ .throwException(new
IllegalArgumentException("Forced"));
+
+ from("direct:race").routeId("race").routePolicy(racePolicy)
+ .process(exchange -> {
+ if
("slow".equals(exchange.getMessage().getBody())) {
+ slowEntered.countDown();
+ assertTrue(slowRelease.await(10,
TimeUnit.SECONDS));
+ }
+ })
+ .filter(body().isEqualTo("fail")).throwException(new
IllegalArgumentException("Forced"));
+ }
+ };
+ }
+
+ /**
+ * 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.
+ */
+ private static final class LatchPolicy extends
ThrottlingExceptionRoutePolicy {
+ private final CountDownLatch halfOpened = new CountDownLatch(1);
+ private final CountDownLatch timerChecking = new CountDownLatch(1);
+ private final CountDownLatch timerMayGo = new CountDownLatch(1);
+ private volatile Thread timerThread;
+
+ LatchPolicy(int threshold, long failureWindow, long halfOpenAfter) {
+ super(threshold, failureWindow, halfOpenAfter, List.of());
+ }
+
+ @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-")) {
+ timerThread = Thread.currentThread();
+ timerChecking.countDown();
+ try {
+ timerMayGo.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return super.isThresholdExceeded();
+ }
+
+ @Override
+ protected void halfOpenCircuit(Route route) {
+ super.halfOpenCircuit(route);
+ halfOpened.countDown();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java
index 352b4537f801..bdf0bb8b031e 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest.java
@@ -110,7 +110,7 @@ public class
ThrottlingExceptionRoutePolicyHalfOpenHandlerSedaTest extends Conte
@Override
public void configure() {
int threshold = 2;
- long failureWindow = 30;
+ long failureWindow = 60000;
long halfOpenAfter = 1000;
ThrottlingExceptionRoutePolicy policy
= new ThrottlingExceptionRoutePolicy(threshold,
failureWindow, halfOpenAfter, null);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java
index d1af59abe46a..091682adcb1e 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenHandlerTest.java
@@ -106,7 +106,7 @@ public class
ThrottlingExceptionRoutePolicyHalfOpenHandlerTest extends ContextTe
@Override
public void configure() {
int threshold = 2;
- long failureWindow = 30;
+ long failureWindow = 60000;
long halfOpenAfter = 250;
ThrottlingExceptionRoutePolicy policy
= new ThrottlingExceptionRoutePolicy(threshold,
failureWindow, halfOpenAfter, null);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenTest.java
index 500d789e8f24..4dfd0004475b 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyHalfOpenTest.java
@@ -106,7 +106,7 @@ public class ThrottlingExceptionRoutePolicyHalfOpenTest
extends ContextTestSuppo
@Override
public void configure() {
int threshold = 2;
- long failureWindow = 30;
+ long failureWindow = 60000;
long halfOpenAfter = 250;
ThrottlingExceptionRoutePolicy policy
= new ThrottlingExceptionRoutePolicy(threshold,
failureWindow, halfOpenAfter, null);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyTest.java
index 3630d9db281d..dbc752ad994f 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyTest.java
@@ -105,7 +105,7 @@ public class ThrottlingExceptionRoutePolicyTest extends
ContextTestSupport {
@Override
public void configure() {
int threshold = 2;
- long failureWindow = 30;
+ long failureWindow = 60000;
long halfOpenAfter = 1000;
ThrottlingExceptionRoutePolicy policy
= new ThrottlingExceptionRoutePolicy(threshold,
failureWindow, halfOpenAfter, null);
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 ab9e81f85e1f..e04c5755e568 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
@@ -16,7 +16,9 @@
*/
package org.apache.camel.throttling;
+import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Deque;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
@@ -48,10 +50,11 @@ import org.slf4j.LoggerFactory;
* database down) then it will stop consuming new messages from the endpoint
by stopping the consumer. The
* implementation is comparable to the Circuit Breaker pattern. After a set
amount of time, it will move to a half open
* state and attempt to determine if the consumer can be started. There are
two ways to determine if a route can be
- * closed after being opened (1) start the consumer and check the failure
threshold (2) call the
- * {@link ThrottlingExceptionHalfOpenHandler} The second option allows a
custom check to be performed without having to
- * take on the possibility of multiple messages from the endpoint. The idea is
that a handler could run a simple test
- * (ie select 1 from dual) to determine if the processes that cause the route
to be open are now available
+ * closed after being opened (1) start the consumer and let the next exchange
decide: a success closes the circuit and a
+ * failure opens it again (2) call the {@link
ThrottlingExceptionHalfOpenHandler} The second option allows a custom
+ * check to be performed without having to take on the possibility of multiple
messages from the endpoint. The idea is
+ * that a handler could run a simple test (ie select 1 from dual) to determine
if the processes that cause the route to
+ * be open are now available
*/
@Metadata(label = "bean",
description = "A throttle based RoutePolicy which is modelled after
the circuit breaker and will stop consuming"
@@ -102,6 +105,10 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
private volatile Timer halfOpenTimer;
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
+ // (not by lock, which is held while the consumer is suspended or resumed)
+ private final Lock failureLock = new ReentrantLock();
+ private final Deque<Long> failureTimes = new ArrayDeque<>();
public ThrottlingExceptionRoutePolicy() {
}
@@ -201,16 +208,19 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
openCircuit(route);
}
} else {
- if (hasFailed(exchange)) {
+ boolean failed = hasFailed(exchange);
+ if (failed) {
// record the failure
failures.incrementAndGet();
- lastFailure = TimeSource.currentTimeMillis();
+ long now = TimeSource.currentTimeMillis();
+ lastFailure = now;
+ recordFailureTime(now);
} else {
success.incrementAndGet();
}
// check for state change
- calculateState(route);
+ calculateState(route, failed);
}
}
@@ -250,7 +260,41 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
return answer;
}
- private void calculateState(Route route) {
+ private void recordFailureTime(long time) {
+ failureLock.lock();
+ try {
+ failureTimes.addLast(time);
+ trimFailureTimes();
+ } finally {
+ failureLock.unlock();
+ }
+ }
+
+ private void clearFailureTimes() {
+ failureLock.lock();
+ try {
+ failureTimes.clear();
+ } finally {
+ failureLock.unlock();
+ }
+ }
+
+ private void trimFailureTimes() {
+ // only the most recent failureThreshold failures decide whether the
threshold is exceeded
+ int max = Math.max(failureThreshold, 1);
+ while (failureTimes.size() > max) {
+ failureTimes.removeFirst();
+ }
+ }
+
+ /**
+ * Calculates the state of the circuit.
+ *
+ * @param route the route
+ * @param exchangeFailed whether the exchange that is done failed, or null
when there is no exchange (the check of
+ * an open circuit after halfOpenAfter)
+ */
+ private void calculateState(Route route, Boolean exchangeFailed) {
// have we reached the failure limit?
boolean failureLimitReached = isThresholdExceeded();
@@ -261,7 +305,14 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
openCircuit(route);
}
} else if (state.get() == STATE_HALF_OPEN) {
- if (failureLimitReached) {
+ // the exchange done in the half open state decides: a failure
opens the circuit again, and a success
+ // closes it. The failures that opened the circuit must not count
against it again: they may still be within
+ // the failure window (with the defaults the circuit is half
opened after 30 seconds, and the window is 60)
+ if (exchangeFailed == null) {
+ // the half open check of the timer, after an exchange that
was done at the same time has already half
+ // opened the circuit: the next exchange decides
+ LOG.debug("Keeping circuit half open (waiting for an
exchange)...");
+ } else if (exchangeFailed) {
LOG.debug("Opening circuit...");
openCircuit(route);
} else {
@@ -300,10 +351,18 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
protected boolean isThresholdExceeded() {
boolean output = false;
logState();
- // failures exceed the threshold
- // AND the last of those failures occurred within window
- if (failures.get() >= failureThreshold && lastFailure >=
TimeSource.currentTimeMillis() - failureWindow) {
- output = true;
+ // at least failureThreshold failures occurred within the window, that
is the oldest of the most recent
+ // failureThreshold failures occurred within the window
+ long since = TimeSource.currentTimeMillis() - failureWindow;
+ failureLock.lock();
+ try {
+ trimFailureTimes();
+ Long oldest = failureTimes.peekFirst();
+ if (failureTimes.size() >= failureThreshold && (oldest != null ?
oldest : lastFailure) >= since) {
+ output = true;
+ }
+ } finally {
+ failureLock.unlock();
}
return output;
@@ -390,6 +449,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
failures.set(0);
success.set(0);
+ clearFailureTimes();
lastFailure = 0;
openedAt = 0;
state.set(STATE_CLOSED);
@@ -444,7 +504,7 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
public void run() {
// cancel the timer of this task only, a newer timer may already
have been scheduled
timer.cancel();
- calculateState(route);
+ calculateState(route, null);
}
}
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 7dcb1fd4ef7f..c91326786302 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
@@ -69,6 +69,19 @@ or payload from the body after a timed out `pollEnrich` must
keep it before the
with the same message id, headers and attachments, so after a timeout it is no
longer of the message type of the
consumer (such as `GenericFileMessage` or `JmsMessage`).
+=== camel-core - ThrottlingExceptionRoutePolicy counts the failures within the
failure window
+
+The circuit of `ThrottlingExceptionRoutePolicy` now opens when
`failureThreshold` failures occurred within the last
+`failureWindow` milliseconds, as documented. Before, it counted every failure
since the circuit last closed (or since
+the route started), and only the last failure had to be within the window, so
failures spread over hours opened the
+circuit on a single new failure. Routes whose failures are spread out
therefore open their circuit less often now; to
+open it sooner, lower `failureThreshold` or raise `failureWindow`.
+
+In the half open state, a successful exchange now closes the circuit, and a
failed exchange opens it again. Before, a
+successful exchange opened the circuit again as long as the failures that
opened it were within `failureWindow`; with
+the defaults (a window of 60 seconds, a half open check after 30 seconds) a
success in the first 30 seconds after the
+half open check therefore kept the circuit open. A `halfOpenHandler` decides
as before.
+
=== 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,
