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 30791176edb5 CAMEL-25300: camel-core - Throttle EIP: the clean of the
total requests throttler must not hand out a second set of permits (#27341)
30791176edb5 is described below
commit 30791176edb5e273304892248c186ea674539339
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 5 22:34:15 2026 +0530
CAMEL-25300: camel-core - Throttle EIP: the clean of the total requests
throttler must not hand out a second set of permits (#27341)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/processor/TotalRequestsThrottler.java | 67 +++++++++-
.../requests/TotalRequestsThrottlerCleanTest.java | 148 +++++++++++++++++++++
.../TotalRequestsThrottlerRateDecreaseTest.java | 90 +++++++++++++
3 files changed, 302 insertions(+), 3 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
index a2c1c2121323..a7f4708bad72 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/TotalRequestsThrottler.java
@@ -59,7 +59,6 @@ public class TotalRequestsThrottler extends AbstractThrottler
{
private static final Logger LOG =
LoggerFactory.getLogger(TotalRequestsThrottler.class);
private volatile long timePeriodMillis;
- private final long cleanPeriodMillis;
private final Expression correlationExpression;
private final Map<String, ThrottlingState> states = new
ConcurrentHashMap<>();
@@ -73,7 +72,6 @@ public class TotalRequestsThrottler extends AbstractThrottler
{
throw new IllegalArgumentException("TimePeriodMillis should be a
positive number, was: " + timePeriodMillis);
}
this.timePeriodMillis = timePeriodMillis;
- this.cleanPeriodMillis = timePeriodMillis * 10;
this.correlationExpression = correlation;
}
@@ -101,6 +99,13 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
throttlingState.calculateAndSetMaxRequestsPerPeriod(exchange);
ThrottlePermit permit = throttlingState.poll();
+ while (permit != null && throttlingState.isRemoved()) {
+ // the state was cleaned after the exchange looked it up, so
the permit does not count:
+ // return it, and take a permit from the state that replaced it
+ throttlingState.returnPermit(permit);
+ throttlingState = throttlingState.currentState(exchange);
+ permit = throttlingState.poll();
+ }
if (permit == null) {
if (isRejectExecution()) {
@@ -126,6 +131,11 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
start = System.currentTimeMillis();
}
permit = throttlingState.take();
+ while (throttlingState.isRemoved()) {
+ throttlingState.returnPermit(permit);
+ throttlingState =
throttlingState.currentState(exchange);
+ permit = throttlingState.take();
+ }
if (LOG.isTraceEnabled()) {
elapsed = System.currentTimeMillis() - start;
}
@@ -226,6 +236,11 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
private final DelayQueue<ThrottlePermit> delayQueue = new
DelayQueue<>();
private final AtomicReference<ScheduledFuture<?>> cleanFuture = new
AtomicReference<>();
private volatile int throttleRate;
+ // not the lock above: a decrease of the throttle rate holds that lock
while it waits for permits, which an
+ // exchange that took a permit returns only after it checked whether
the state was removed
+ private final Lock removedLock = new ReentrantLock();
+ // guarded by removedLock
+ private boolean removed;
ThrottlingState(String key) {
this.key = key;
@@ -247,8 +262,50 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
return delayQueue.take();
}
+ /**
+ * Removes this state if all its permits are returned (no exchange
took a permit it has not returned yet). Only
+ * this state is removed, and not a state that has already replaced it.
+ */
public void clean() {
- states.remove(key);
+ states.computeIfPresent(key, (k, s) -> s == this &&
markRemovedIfUnused() ? null : s);
+ }
+
+ private boolean markRemovedIfUnused() {
+ removedLock.lock();
+ try {
+ if (delayQueue.size() >= throttleRate) {
+ removed = true;
+ }
+ return removed;
+ } finally {
+ removedLock.unlock();
+ }
+ }
+
+ /**
+ * Whether this state was removed by {@link #clean()} after the
exchange looked it up, in which case a permit
+ * taken from it does not count, and must be taken from the state that
replaced it instead.
+ */
+ private boolean isRemoved() {
+ removedLock.lock();
+ try {
+ return removed;
+ } finally {
+ removedLock.unlock();
+ }
+ }
+
+ private ThrottlingState currentState(Exchange exchange) throws
Exception {
+ ThrottlingState answer = states.computeIfAbsent(key,
ThrottlingState::new);
+ answer.calculateAndSetMaxRequestsPerPeriod(exchange);
+ return answer;
+ }
+
+ /**
+ * Puts a permit back as it was, without a new delay.
+ */
+ private void returnPermit(ThrottlePermit permit) {
+ delayQueue.put(permit);
}
/**
@@ -258,6 +315,10 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
permit.setDelayMs(getTimePeriodMillis());
delayQueue.put(permit);
try {
+ // clean up 10 periods after the last permit was returned
(with the current period, which can be
+ // changed at runtime), when no permit is delayed any longer
+ long period = getTimePeriodMillis();
+ long cleanPeriodMillis = period > Long.MAX_VALUE / 10 ?
Long.MAX_VALUE : period * 10;
ScheduledFuture<?> next = asyncExecutor.schedule(this::clean,
cleanPeriodMillis, TimeUnit.MILLISECONDS);
ScheduledFuture<?> prev = cleanFuture.getAndSet(next);
if (prev != null) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/TotalRequestsThrottlerCleanTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/TotalRequestsThrottlerCleanTest.java
new file mode 100644
index 000000000000..ebd7c98755a5
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/TotalRequestsThrottlerCleanTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.requests;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.ThrottlerRejectedExecutionException;
+import org.apache.camel.processor.TotalRequestsThrottler;
+import org.apache.camel.support.ExpressionAdapter;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * The throttler cleans up its state some time after a permit was last
returned. An exchange that looked up the state
+ * before the clean must not take a permit from the removed state, as the next
exchanges get the permits of the state
+ * that replaces it, and more exchanges than allowed would pass in the time
period.
+ */
+public class TotalRequestsThrottlerCleanTest extends ContextTestSupport {
+
+ private final CapturingExecutor executor = new CapturingExecutor();
+
+ @AfterEach
+ public void shutdownExecutor() {
+ executor.shutdownNow();
+ }
+
+ @Test
+ public void testCleanAfterExchangeLookedUpState() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("first", "race",
"second");
+
+ // takes a permit, which schedules the clean
+ template.sendBody("direct:start", "first");
+ // the clean runs after race looked up the state, but before it takes
a permit
+ template.sendBody("direct:start", "race");
+
+ // race took its permit from the state that replaced the cleaned one,
so that state hands out only one more
+ // permit, not a second full set next to the permit race took. The
clean runs here while the permit of first
+ // is still delayed (in practice it runs 10 periods later), so first,
race and second all pass: this test is
+ // about the permits of the replacement state, not about the
per-period limit
+ template.sendBody("direct:start", "second");
+ Exception e = assertThrows(CamelExecutionException.class,
+ () -> template.sendBody("direct:start", "rejected"));
+ assertInstanceOf(ThrottlerRejectedExecutionException.class,
e.getCause());
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testCleanRemovesUnusedState() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("first");
+
+ template.sendBody("direct:start", "first");
+ assertMockEndpointsSatisfied();
+
+ TotalRequestsThrottler throttler = context.getProcessor("throttler",
TotalRequestsThrottler.class);
+ assertEquals(2, throttler.getCurrentMaximumRequests());
+
+ executor.runScheduled();
+ assertEquals(0, throttler.getCurrentMaximumRequests());
+ }
+
+ @Test
+ public void testCleanFollowsTimePeriod() throws Exception {
+ TotalRequestsThrottler throttler = context.getProcessor("throttler",
TotalRequestsThrottler.class);
+ // such as from JMX
+ throttler.setTimePeriodMillis(TimeUnit.HOURS.toMillis(2));
+
+ template.sendBody("direct:start", "first");
+
+ // the clean runs 10 periods after the last permit was returned, so no
permit is still delayed then
+ assertEquals(List.of(TimeUnit.HOURS.toMillis(20)), executor.delays);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .throttle(new ExpressionAdapter() {
+ @Override
+ public Object evaluate(Exchange exchange) {
+ if
("race".equals(exchange.getMessage().getBody())) {
+ // the throttler evaluates this after
looking up its state for the exchange
+ executor.runScheduled();
+ }
+ return 2;
+ }
+
}).timePeriodMillis(TimeUnit.HOURS.toMillis(1)).rejectExecution(true).executorService(executor)
+ .id("throttler")
+ .to("mock:result");
+ }
+ };
+ }
+
+ /**
+ * Keeps the scheduled clean tasks so the test decides when they run.
+ */
+ private static final class CapturingExecutor extends
ScheduledThreadPoolExecutor {
+ private final List<Runnable> scheduled = new CopyOnWriteArrayList<>();
+ private final List<Long> delays = new CopyOnWriteArrayList<>();
+
+ CapturingExecutor() {
+ super(1);
+ setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ }
+
+ @Override
+ public ScheduledFuture<?> schedule(Runnable command, long delay,
TimeUnit unit) {
+ scheduled.add(command);
+ delays.add(unit.toMillis(delay));
+ return super.schedule(command, 1, TimeUnit.DAYS);
+ }
+
+ void runScheduled() {
+ for (Runnable task : scheduled) {
+ task.run();
+ }
+ scheduled.clear();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/TotalRequestsThrottlerRateDecreaseTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/TotalRequestsThrottlerRateDecreaseTest.java
new file mode 100644
index 000000000000..fe9c50b0d687
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/requests/TotalRequestsThrottlerRateDecreaseTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.requests;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+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;
+
+/**
+ * A decrease of the maximum requests waits for permits while other exchanges
wait for permits too: the exchanges that
+ * get the permits must be able to return them, so the decrease completes.
+ */
+public class TotalRequestsThrottlerRateDecreaseTest extends ContextTestSupport
{
+
+ private static final long PERIOD = 2000;
+
+ private final List<Throwable> failures = new CopyOnWriteArrayList<>();
+
+ @Test
+ public void testDecreaseWhileExchangesWaitForPermits() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(5);
+
+ // both permits are taken, and delayed for one period
+ template.sendBodyAndHeader("direct:start", "a", "max", 2);
+ template.sendBodyAndHeader("direct:start", "b", "max", 2);
+
+ CountDownLatch done = new CountDownLatch(3);
+ // two exchanges wait for the delayed permits (the first one gets the
first permit, the second one the next)
+ Thread first = send("c", 2, done);
+ await().atMost(5, TimeUnit.SECONDS).pollInterval(10,
TimeUnit.MILLISECONDS)
+ .until(() -> first.getState() == Thread.State.TIMED_WAITING);
+ Thread second = send("d", 2, done);
+ await().atMost(5, TimeUnit.SECONDS).pollInterval(10,
TimeUnit.MILLISECONDS)
+ .until(() -> second.getState() == Thread.State.WAITING);
+ // then an exchange decreases the maximum to 1, which waits for a
permit to discard
+ send("e", 1, done);
+
+ assertTrue(done.await(20, TimeUnit.SECONDS), "All exchanges should get
through the throttler");
+ assertEquals(List.of(), failures);
+ assertMockEndpointsSatisfied();
+ }
+
+ private Thread send(String body, int max, CountDownLatch done) {
+ Thread thread = new Thread(() -> {
+ try {
+ template.sendBodyAndHeader("direct:start", body, "max", max);
+ } catch (Exception e) {
+ failures.add(e);
+ } finally {
+ done.countDown();
+ }
+ }, "send-" + body);
+ thread.setDaemon(true);
+ thread.start();
+ return thread;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").throttle(header("max")).timePeriodMillis(PERIOD).to("mock:result");
+ }
+ };
+ }
+}
