This is an automated email from the ASF dual-hosted git repository.
gnodet 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 bd4d2ba20d6a CAMEL-24903: Fix flaky
ThrottlingExceptionRoutePolicyOpenViaConfigTest (#26799)
bd4d2ba20d6a is described below
commit bd4d2ba20d6a268b66949712d00073069a8e333d
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Sep 24 16:18:27 2026 +0200
CAMEL-24903: Fix flaky ThrottlingExceptionRoutePolicyOpenViaConfigTest
(#26799)
* CAMEL-24903: Fix flaky ThrottlingExceptionRoutePolicyOpenViaConfigTest
Root cause: with concurrentConsumers=20, a round-1 onExchangeDone()
callback could fire after policy.setKeepOpen(true) was called, seeing
keepOpenBool=true and calling openCircuit() to suspend the consumer
BEFORE the 'MessageTrigger' message was picked up from the SEDA queue.
This caused the expectedMessageCount(6) assertion to fail with count 5.
Fix (two-pronged):
1. ThrottlingExceptionRoutePolicy.setKeepOpen(true) now immediately
opens the circuit via openCircuit(route) rather than waiting for the
next onExchangeDone() callback. This makes the keepOpen toggle
synchronous and deterministic.
2. Test reordering: 'MessageTrigger' is now sent and awaited BEFORE
setKeepOpen(true) is called, ensuring it is consumed while the
circuit is still closed. setKeepOpen(true) then suspends the consumer
immediately so no subsequent messages flow through.
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
* CAMEL-24903: Address review feedback
- Move state.get() != STATE_OPEN guard inside openCircuit() under the
lock so concurrent callers are idempotent and spurious timer chains
cannot accumulate (davsclaus item 1)
- Remove now-redundant outer guard from setKeepOpen(); add Javadoc
documenting synchronous-open semantics and single-route ownership
(davsclaus items 1 & 2)
- Add upgrade-guide entry for the behaviour change in setKeepOpen(true)
(davsclaus item 3)
- Add setAssertPeriod(500) to the round-2 keepOpen assertion so it
actually verifies no extra messages arrive while the circuit is open
(davsclaus blue nit)
---------
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
...tlingExceptionRoutePolicyOpenViaConfigTest.java | 18 ++++++++++++------
.../throttling/ThrottlingExceptionRoutePolicy.java | 22 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 9 +++++++++
3 files changed, 43 insertions(+), 6 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyOpenViaConfigTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyOpenViaConfigTest.java
index 75d7b7b81abc..46ea90090390 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyOpenViaConfigTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/ThrottlingExceptionRoutePolicyOpenViaConfigTest.java
@@ -69,12 +69,16 @@ class ThrottlingExceptionRoutePolicyOpenViaConfigTest
extends ContextTestSupport
result.expectedMessageCount(size);
MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
- // set keepOpen to true
- policy.setKeepOpen(true);
-
- // trigger opening circuit
- // by sending another message
+ // send the trigger message while the circuit is still closed so it is
+ // guaranteed to be consumed before we open the circuit
template.sendBody(url, "MessageTrigger");
+ result.expectedMessageCount(size + 1);
+ MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+
+ // toggle keepOpen to true: setKeepOpen() immediately suspends the
+ // consumer so no late onExchangeDone() callback can race against the
+ // trigger message (CAMEL-24903)
+ policy.setKeepOpen(true);
// wait for the circuit to open (consumer suspended)
await().atMost(10, TimeUnit.SECONDS).until(consumer::isSuspended);
@@ -85,9 +89,11 @@ class ThrottlingExceptionRoutePolicyOpenViaConfigTest
extends ContextTestSupport
template.sendBody(url, "MessageRound2 " + i);
}
- // should not close b/c keepOpen is true
+ // should not close b/c keepOpen is true; use assertPeriod to verify
no extra messages arrive
+ result.setAssertPeriod(500);
result.expectedMessageCount(size + 1);
MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ result.setAssertPeriod(0);
// set keepOpen to false
policy.setKeepOpen(false);
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 1254990c99fe..e453f4be20b3 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
@@ -308,6 +308,9 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
protected void openCircuit(Route route) {
try {
lock.lock();
+ if (state.get() == STATE_OPEN) {
+ return;
+ }
suspendOrStopConsumer(route.getConsumer());
state.set(STATE_OPEN);
openedAt = System.currentTimeMillis();
@@ -413,8 +416,27 @@ public class ThrottlingExceptionRoutePolicy extends
RoutePolicySupport implement
return this.keepOpenBool.get();
}
+ /**
+ * Sets whether to keep the circuit breaker always open (never closes).
This is only intended for development and
+ * testing purposes.
+ * <p>
+ * When set to {@code true}, the circuit is opened immediately and
synchronously (the consumer is suspended before
+ * this method returns). Setting it back to {@code false} is deferred: the
half-open timer will attempt to close the
+ * circuit on its next tick (after {@link #getHalfOpenAfter()}
milliseconds).
+ * <p>
+ * Note: a single {@link ThrottlingExceptionRoutePolicy} instance is
designed for one route. When attached to
+ * multiple routes, only the last route bound via {@link #setRoute(Route)}
is acted upon by this setter.
+ */
public void setKeepOpen(boolean keepOpen) {
this.keepOpenBool.set(keepOpen);
+ // Immediately act on the new value so callers do not have to send a
+ // trigger message and wait for the next onExchangeDone() callback:
+ // - keepOpen=true → open the circuit right away (openCircuit is
idempotent under its lock)
+ // - keepOpen=false → the half-open timer will close the circuit on
its next tick
+ if (keepOpen && route != null) {
+ LOG.debug("Opening circuit (keepOpen set to true)");
+ openCircuit(route);
+ }
}
public int getFailureThreshold() {
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index d23eb1f20856..f70b258eedc1 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -2812,3 +2812,12 @@ producer were never stopped.
With `parallelProcessing`, a recipient whose task had not started yet when the
Recipient List completed is now
skipped instead of being sent to afterwards. As before, recipients that had
already started keep running.
+
+== ThrottlingExceptionRoutePolicy
+
+`ThrottlingExceptionRoutePolicy.setKeepOpen(true)` now opens the circuit
immediately and synchronously (the consumer
+is suspended before the setter returns) rather than waiting for the next
`onExchangeDone()` callback.
+This is the intended behaviour and makes the `keepOpen` toggle deterministic,
but callers that previously
+relied on the setter being inert until an exchange arrived must be aware of
the change.
+Setting `keepOpen` back to `false` remains deferred: the half-open timer
attempts to close the circuit on
+its next tick (after `halfOpenAfter` milliseconds, default 30 s).
