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 75bea5f705b1 CAMEL-26313: camel-core - ThrottlingExceptionRoutePolicy:
raising halfOpenAfter while the circuit is open no longer leaves the route
suspended (#28574)
75bea5f705b1 is described below
commit 75bea5f705b13df07a1efbad348cc98c21556552
Author: Vignesh Nayak Manel <[email protected]>
AuthorDate: Wed Oct 28 11:40:20 2026 +0530
CAMEL-26313: camel-core - ThrottlingExceptionRoutePolicy: raising
halfOpenAfter while the circuit is open no longer leaves the route suspended
(#28574)
* CAMEL-26313: camel-core - ThrottlingExceptionRoutePolicy: raising
halfOpenAfter while the circuit is open no longer leaves the route suspended
The half open timer is scheduled with the value of halfOpenAfter when the
circuit opens. If halfOpenAfter is raised while the circuit is open (it is
writable over JMX), the timer fires before the new value, and calculateState
took the "time not elapsed" branch, which scheduled nothing. The consumer is
suspended, so no exchange came to check again, and the route stayed
suspended
until it was restarted.
In that branch, schedule the timer again for the time that is left.
halfOpenAfter is read once, so a value lowered at the same time cannot give
a negative delay.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-26313: camel-core - ThrottlingExceptionRoutePolicy: only the half
open timer schedules the check again
An exchange still in flight when the circuit opened is done while the
circuit is open, and took the "time not elapsed" branch too: each one
created a new Timer (a new thread) and cancelled the previous one. Only
the check of the timer can leave the route suspended, so only that path
schedules the check again.
addHalfOpenTimer(Route, long) is now protected, so a subclass that
customises the scheduling also handles the timer scheduled again.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
...gExceptionRoutePolicyInflightWhileOpenTest.java | 102 +++++++++++++++++++++
...ExceptionRoutePolicyRaiseHalfOpenAfterTest.java | 72 +++++++++++++++
.../throttling/ThrottlingExceptionRoutePolicy.java | 19 +++-
3 files changed, 191 insertions(+), 2 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyInflightWhileOpenTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyInflightWhileOpenTest.java
new file mode 100644
index 000000000000..b716ea2e15a6
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyInflightWhileOpenTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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.support.service.ServiceSupport;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
+import org.junit.jupiter.api.BeforeEach;
+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.assertTrue;
+
+/**
+ * An exchange that is done while the circuit is open must not schedule the
half open check again: only the timer that
+ * was scheduled when the circuit opened (each timer has its own thread).
+ */
+class ThrottlingExceptionRoutePolicyInflightWhileOpenTest extends
ContextTestSupport {
+
+ private final AtomicInteger timers = new AtomicInteger();
+ private final AtomicInteger done = new AtomicInteger();
+ private final CountDownLatch slowStarted = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ private final ThrottlingExceptionRoutePolicy policy = new
ThrottlingExceptionRoutePolicy(1, 60000, 60000, null) {
+ @Override
+ protected void addHalfOpenTimer(Route route, long delay) {
+ timers.incrementAndGet();
+ super.addHalfOpenTimer(route, delay);
+ }
+
+ @Override
+ public void onExchangeDone(Route route, Exchange exchange) {
+ super.onExchangeDone(route, exchange);
+ done.incrementAndGet();
+ }
+ };
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ context.getShutdownStrategy().setTimeout(1);
+ }
+
+ @Test
+ void testExchangeDoneWhileOpenDoesNotScheduleTheHalfOpenCheckAgain()
throws Exception {
+ template.sendBody("seda:start", "slow");
+ assertTrue(slowStarted.await(10, TimeUnit.SECONDS), "the slow exchange
should be in flight");
+
+ // opens the circuit, which schedules the half open check
+ template.sendBody("seda:start", "fail");
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("start").getConsumer();
+ await().atMost(10, TimeUnit.SECONDS).until(consumer::isSuspended);
+ assertEquals(1, timers.get(), "opening the circuit schedules the half
open check");
+
+ // the slow exchange is done while the circuit is open
+ release.countDown();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> done.get() == 2);
+ assertEquals(1, timers.get(), "an exchange done while the circuit is
open must not schedule another check");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:start?concurrentConsumers=2").routeId("start").routePolicy(policy)
+ .process(e -> {
+ if
("slow".equals(e.getMessage().getBody(String.class))) {
+ slowStarted.countDown();
+ release.await(10, TimeUnit.SECONDS);
+ } else {
+ throw new ThrottlingException("boom");
+ }
+ });
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyRaiseHalfOpenAfterTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyRaiseHalfOpenAfterTest.java
new file mode 100644
index 000000000000..cd275ae7b2a1
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyRaiseHalfOpenAfterTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.camel.throttling.ThrottlingExceptionRoutePolicy;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Raising halfOpenAfter while the circuit is open delays the half open check,
it must not leave the route suspended.
+ */
+class ThrottlingExceptionRoutePolicyRaiseHalfOpenAfterTest extends
ContextTestSupport {
+
+ private final ThrottlingExceptionRoutePolicy policy = new
ThrottlingExceptionRoutePolicy(1, 60000, 2000, null);
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ context.getShutdownStrategy().setTimeout(1);
+ }
+
+ @Test
+ void testRaisedHalfOpenAfterHalfOpensTheCircuit() throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.whenExchangeReceived(1, e -> e.setException(new
ThrottlingException("boom")));
+ result.expectedBodiesReceived("fail", "ok");
+
+ template.sendBody("seda:start", "fail");
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("start").getConsumer();
+ await().atMost(10, TimeUnit.SECONDS).until(consumer::isSuspended);
+
+ // as over JMX: the half open timer is due after the old value, before
the new one
+ policy.setHalfOpenAfter(3000);
+
+ await().atMost(10, TimeUnit.SECONDS).until(consumer::isStarted);
+ template.sendBody("seda:start", "ok");
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:start").routeId("start").routePolicy(policy).to("mock:result");
+ }
+ };
+ }
+}
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 e04c5755e568..255bcd7be538 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
@@ -322,7 +322,9 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
} else if (state.get() == STATE_OPEN) {
if (!keepOpenBool.get()) {
long elapsedTimeSinceOpened = TimeSource.currentTimeMillis() -
openedAt;
- if (halfOpenAfter <= elapsedTimeSinceOpened) {
+ // read once, as halfOpenAfter can be changed at runtime (such
as over JMX)
+ long openFor = halfOpenAfter;
+ if (openFor <= elapsedTimeSinceOpened) {
LOG.debug("Checking an open circuit...");
if (halfOpenHandler != null) {
if (halfOpenHandler.isReadyToBeClosed()) {
@@ -337,7 +339,13 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
LOG.debug("Half opening circuit...");
halfOpenCircuit(route);
}
+ } else if (exchangeFailed == null) {
+ LOG.debug("Keeping circuit open (time not elapsed)...");
+ // the half open timer fired before halfOpenAfter, as it
was raised after the timer was scheduled:
+ // the consumer is suspended, so no exchange comes to
check again. Check when the time is up
+ addHalfOpenTimer(route, openFor - elapsedTimeSinceOpened);
} else {
+ // an exchange still in flight when the circuit opened:
the half open timer does the check
LOG.debug("Keeping circuit open (time not elapsed)...");
}
} else {
@@ -407,11 +415,18 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
}
protected void addHalfOpenTimer(Route route) {
+ addHalfOpenTimer(route, halfOpenAfter);
+ }
+
+ /**
+ * Schedules the half open check after the given delay, replacing the one
that is scheduled.
+ */
+ protected void addHalfOpenTimer(Route route, long delay) {
lock.lock();
try {
Timer previous = halfOpenTimer;
Timer timer = new Timer();
- timer.schedule(new HalfOpenTask(route, timer), halfOpenAfter);
+ timer.schedule(new HalfOpenTask(route, timer), delay);
halfOpenTimer = timer;
if (previous != null) {
// only one half open check at a time (each timer has its own
thread)
