This is an automated email from the ASF dual-hosted git repository.
apupier 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 56c40ea0035b CAMEL-24926: camel-core - Concurrent requests throttler
should not let more exchanges in than allowed after its state is cleaned
56c40ea0035b is described below
commit 56c40ea0035b5dc153244ec29b2fe12ccfee8962
Author: smjain <[email protected]>
AuthorDate: Wed Sep 23 16:58:08 2026 +0530
CAMEL-24926: camel-core - Concurrent requests throttler should not let more
exchanges in than allowed after its state is cleaned
ConcurrentRequestsThrottler removes the state of a key (the semaphore) 10
seconds after a permit was returned, to not keep states for correlation keys
that are no longer used. It removed the state even if another exchange had
taken a permit in the meantime and was still being processed. The next
exchange then created a new state with all the permits, so with
throttle(1).concurrentRequestsMode() two exchanges were processed at the
same
time as soon as one of them took longer than 10 seconds.
The state is now only removed when no permit is taken and nobody waits for
one. An exchange that looked up the state just before it was removed gives
the permit back and takes one from the state that replaces it.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/ConcurrentRequestsThrottler.java | 55 ++++++-
.../ConcurrentRequestsThrottlerCleanTest.java | 173 +++++++++++++++++++++
2 files changed, 224 insertions(+), 4 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
index 1885f5abd8bc..afd90ce238e0 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/ConcurrentRequestsThrottler.java
@@ -147,7 +147,7 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
}
private static void doThrottle(Exchange exchange, ThrottlingState
throttlingState, State state, long queuedStart)
- throws InterruptedException {
+ throws Exception {
// block waiting for a permit
long start = 0;
long elapsed = 0;
@@ -222,6 +222,8 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
private final AtomicReference<ScheduledFuture<?>> cleanFuture = new
AtomicReference<>();
private volatile int throttleRate;
private final WrappedSemaphore semaphore;
+ // guarded by lock
+ private boolean removed;
ThrottlingState(String key) {
this.key = key;
@@ -232,20 +234,65 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
return throttleRate;
}
+ /**
+ * Removes this state if no permit is taken and nobody is waiting for
one. A state that is in use must be kept,
+ * otherwise the next exchange would create a new state with all the
permits, and more exchanges than allowed
+ * would be processed at the same time.
+ */
public void clean() {
- states.remove(key);
+ states.computeIfPresent(key, (k, s) -> s == this &&
markRemovedIfUnused() ? null : s);
+ }
+
+ private boolean markRemovedIfUnused() {
+ lock.lock();
+ try {
+ if (semaphore.availablePermits() >= throttleRate &&
!semaphore.hasQueuedThreads()) {
+ removed = true;
+ }
+ return removed;
+ } finally {
+ lock.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() {
+ lock.lock();
+ try {
+ return removed;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private ThrottlingState currentState(Exchange exchange) throws
Exception {
+ ThrottlingState answer = states.computeIfAbsent(key,
ThrottlingState::new);
+ answer.calculateAndSetMaxConcurrentRequestsExpression(exchange);
+ return answer;
}
- public boolean tryAcquire(Exchange exchange) {
+ public boolean tryAcquire(Exchange exchange) throws Exception {
boolean acquired = semaphore.tryAcquire();
if (acquired) {
+ if (isRemoved()) {
+ semaphore.release();
+ return currentState(exchange).tryAcquire(exchange);
+ }
addSynchronization(exchange);
}
return acquired;
}
- public void acquire(Exchange exchange) throws InterruptedException {
+ public void acquire(Exchange exchange) throws Exception {
semaphore.acquire();
+ if (isRemoved()) {
+ semaphore.release();
+ currentState(exchange).acquire(exchange);
+ return;
+ }
addSynchronization(exchange);
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/throttle/concurrent/ConcurrentRequestsThrottlerCleanTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/concurrent/ConcurrentRequestsThrottlerCleanTest.java
new file mode 100644
index 000000000000..cddaed78ad0a
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/throttle/concurrent/ConcurrentRequestsThrottlerCleanTest.java
@@ -0,0 +1,173 @@
+/*
+ * 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.concurrent;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+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.ConcurrentRequestsThrottler;
+import org.apache.camel.processor.ThrottlerRejectedExecutionException;
+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;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The throttler cleans up its state some time after the last permit was
returned. That must not happen while an
+ * exchange that took a permit later is still being processed, or the next
exchange gets a fresh set of permits.
+ */
+public class ConcurrentRequestsThrottlerCleanTest extends ContextTestSupport {
+
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final CapturingExecutor executor = new CapturingExecutor();
+
+ @AfterEach
+ public void shutdownExecutor() {
+ executor.shutdownNow();
+ }
+
+ @Test
+ public void testCleanDoesNotRemoveStateInUse() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("fast", "slow",
"after");
+
+ // takes and returns the only permit, which schedules the clean
+ template.sendBody("direct:start", "fast");
+
+ Future<Object> slow = template.asyncSendBody("direct:start", "slow");
+ assertTrue(entered.await(10, TimeUnit.SECONDS));
+
+ // the clean runs while slow holds the only permit
+ executor.runScheduled();
+
+ // so there is still no permit for another exchange
+ Exception e = assertThrows(CamelExecutionException.class,
+ () -> template.sendBody("direct:start", "rejected"));
+ assertInstanceOf(ThrottlerRejectedExecutionException.class,
e.getCause());
+
+ release.countDown();
+ slow.get(10, TimeUnit.SECONDS);
+
+ template.sendBody("direct:start", "after");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testCleanAfterExchangeLookedUpState() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("fast", "race",
"after");
+
+ // takes and returns the only permit, which schedules the clean
+ template.sendBody("direct:start", "fast");
+
+ // the clean runs after race looked up the state, but before it takes
the permit
+ Future<Object> race = template.asyncSendBody("direct:start", "race");
+ assertTrue(entered.await(10, TimeUnit.SECONDS));
+
+ // race holds the only permit
+ Exception e = assertThrows(CamelExecutionException.class,
+ () -> template.sendBody("direct:start", "rejected"));
+ assertInstanceOf(ThrottlerRejectedExecutionException.class,
e.getCause());
+
+ release.countDown();
+ race.get(10, TimeUnit.SECONDS);
+
+ template.sendBody("direct:start", "after");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ public void testCleanRemovesUnusedState() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("fast");
+
+ template.sendBody("direct:start", "fast");
+ assertMockEndpointsSatisfied();
+
+ ConcurrentRequestsThrottler throttler =
context.getProcessor("throttler", ConcurrentRequestsThrottler.class);
+ assertEquals(1, throttler.getCurrentMaximumRequests());
+
+ executor.runScheduled();
+ assertEquals(0, throttler.getCurrentMaximumRequests());
+ }
+
+ @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 1;
+ }
+
}).concurrentRequestsMode().rejectExecution(true).executorService(executor).id("throttler")
+ .process(e -> {
+ Object body = e.getMessage().getBody();
+ if ("slow".equals(body) || "race".equals(body)) {
+ entered.countDown();
+ release.await(10, TimeUnit.SECONDS);
+ }
+ })
+ .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<>();
+
+ CapturingExecutor() {
+ super(1);
+ setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ }
+
+ @Override
+ public ScheduledFuture<?> schedule(Runnable command, long delay,
TimeUnit unit) {
+ scheduled.add(command);
+ return super.schedule(command, 1, TimeUnit.DAYS);
+ }
+
+ void runScheduled() {
+ for (Runnable task : scheduled) {
+ task.run();
+ }
+ scheduled.clear();
+ }
+ }
+}
