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 a9ca7702ba76 CAMEL-26472: camel-core - Throttle EIP: do not reschedule
the clean of the throttling state on every exchange (#28816)
a9ca7702ba76 is described below
commit a9ca7702ba7693548bace7ebad3ed2e9800939d7
Author: Adriano Machado <[email protected]>
AuthorDate: Mon Nov 2 01:30:33 2026 -0500
CAMEL-26472: camel-core - Throttle EIP: do not reschedule the clean of the
throttling state on every exchange (#28816)
…f the throttling state on every exchange
Both throttler modes scheduled a new clean and cancelled the previous one
for every exchange. The pending clean now stays while at least a fifth of
the clean period is left. In total requests mode that is 2 periods, so it
still runs after the delay of the permit returned, which matters as a clean
removes a state whose permits are all returned even if they are delayed
(CAMEL-25300). A clean now runs 8 to 10 periods after the last permit was
returned, instead of exactly 10.
ThrottlePermit.compareTo compares the scheduled times, without reading the
clock twice.
_Claude Code on behalf of Adriano Machado (@ammachado)_
_This was generated by an AI agent and may contain inaccuracies.
Please verify before relying on it._
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/processor/AbstractThrottler.java | 20 ++++++++++++
.../processor/ConcurrentRequestsThrottler.java | 7 ++--
.../camel/processor/TotalRequestsThrottler.java | 12 +++----
.../ConcurrentRequestsThrottlerCleanTest.java | 23 ++++++++++++-
.../requests/TotalRequestsThrottlerCleanTest.java | 38 ++++++++++++++++++++--
5 files changed, 85 insertions(+), 15 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
index 3a7bb0b973cd..892278ca723f 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/AbstractThrottler.java
@@ -17,6 +17,9 @@
package org.apache.camel.processor;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
@@ -62,6 +65,23 @@ public abstract class AbstractThrottler extends
BaseProcessorSupport
this.maxRequestsExpression =
ObjectHelper.notNull(maxRequestsExpression, "maxConcurrentRequestsExpression");
}
+ /**
+ * Schedules a clean of a throttling state, unless the pending one can
stay. This is called for every exchange, and
+ * scheduling and cancelling takes the executor queue lock about three
times. The pending clean stays while at least
+ * a fifth of the clean period is left, so it may run up to a fifth
earlier than the clean period after this call.
+ */
+ void scheduleClean(AtomicReference<ScheduledFuture<?>> cleanFuture,
Runnable clean, long cleanPeriodMillis) {
+ ScheduledFuture<?> pending = cleanFuture.get();
+ if (pending != null && pending.getDelay(TimeUnit.MILLISECONDS) >=
cleanPeriodMillis / 5) {
+ return;
+ }
+ ScheduledFuture<?> next = asyncExecutor.schedule(clean,
cleanPeriodMillis, TimeUnit.MILLISECONDS);
+ ScheduledFuture<?> prev = cleanFuture.getAndSet(next);
+ if (prev != null) {
+ prev.cancel(false);
+ }
+ }
+
protected static boolean handleInterrupt(
Exchange exchange, AsyncCallback callback, InterruptedException e,
boolean doneSync) {
// determine if we can still run, or the camel context is forcing a
shutdown
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 82e7033650ba..ba5e9247b542 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
@@ -223,6 +223,7 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
private final String key;
private final Lock lock = new ReentrantLock();
private final AtomicReference<ScheduledFuture<?>> cleanFuture = new
AtomicReference<>();
+ private final Runnable cleanTask = this::clean;
private volatile int throttleRate;
private final WrappedSemaphore semaphore;
// guarded by lock
@@ -319,11 +320,7 @@ public class ConcurrentRequestsThrottler extends
AbstractThrottler {
public void release(final Exchange exchange) {
semaphore.release();
try {
- ScheduledFuture<?> next = asyncExecutor.schedule(this::clean,
CLEAN_PERIOD, TimeUnit.MILLISECONDS);
- ScheduledFuture<?> prev = cleanFuture.getAndSet(next);
- if (prev != null) {
- prev.cancel(false);
- }
+ scheduleClean(cleanFuture, cleanTask, CLEAN_PERIOD);
if (LOG.isTraceEnabled()) {
LOG.trace("Permit released, for exchangeId: {}",
exchange.getExchangeId());
}
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 f949915439ad..a7181559a44a 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
@@ -237,6 +237,7 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
private final Lock lock = new ReentrantLock();
private final DelayQueue<ThrottlePermit> delayQueue = new
DelayQueue<>();
private final AtomicReference<ScheduledFuture<?>> cleanFuture = new
AtomicReference<>();
+ private final Runnable cleanTask = this::clean;
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
@@ -317,15 +318,12 @@ 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
+ // clean up 8 to 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) {
- prev.cancel(false);
- }
+ // a clean that stays pending runs at least 2 periods from
now, so after the delay of this permit
+ scheduleClean(cleanFuture, cleanTask, cleanPeriodMillis);
// try and incur the least amount of overhead while releasing
permits back to the queue
if (LOG.isTraceEnabled()) {
LOG.trace("Permit released, for exchangeId: {}",
exchange.getExchangeId());
@@ -424,7 +422,7 @@ public class TotalRequestsThrottler extends
AbstractThrottler {
@Override
public int compareTo(final Delayed o) {
- return Long.compare(getDelay(TimeUnit.MILLISECONDS),
o.getDelay(TimeUnit.MILLISECONDS));
+ return Long.compare(scheduledTime, ((ThrottlePermit)
o).scheduledTime);
}
}
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
index cddaed78ad0a..60f36f91d469 100644
---
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
@@ -118,6 +118,26 @@ public class ConcurrentRequestsThrottlerCleanTest extends
ContextTestSupport {
assertEquals(0, throttler.getCurrentMaximumRequests());
}
+ @Test
+ public void testCleanStaysScheduledWhileFarFromFiring() throws Exception {
+ template.sendBody("direct:start", "first");
+ template.sendBody("direct:start", "second");
+
+ // scheduling and cancelling per exchange would take the executor
queue lock about three times
+ assertEquals(1, executor.scheduled.size());
+ }
+
+ @Test
+ public void testCleanIsRescheduledWhenCloseToFiring() throws Exception {
+ // otherwise a state still in use when the pending clean runs would
never be cleaned
+ executor.pendingDelayMillis = TimeUnit.SECONDS.toMillis(1);
+
+ template.sendBody("direct:start", "first");
+ template.sendBody("direct:start", "second");
+
+ assertEquals(2, executor.scheduled.size());
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -151,6 +171,7 @@ public class ConcurrentRequestsThrottlerCleanTest extends
ContextTestSupport {
*/
private static final class CapturingExecutor extends
ScheduledThreadPoolExecutor {
private final List<Runnable> scheduled = new CopyOnWriteArrayList<>();
+ private volatile long pendingDelayMillis = TimeUnit.DAYS.toMillis(1);
CapturingExecutor() {
super(1);
@@ -160,7 +181,7 @@ public class ConcurrentRequestsThrottlerCleanTest extends
ContextTestSupport {
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay,
TimeUnit unit) {
scheduled.add(command);
- return super.schedule(command, 1, TimeUnit.DAYS);
+ return super.schedule(command, pendingDelayMillis,
TimeUnit.MILLISECONDS);
}
void runScheduled() {
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
index ebd7c98755a5..90561c11e7e5 100644
---
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
@@ -61,7 +61,7 @@ public class TotalRequestsThrottlerCleanTest extends
ContextTestSupport {
// 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
+ // is still delayed (in practice it runs 8 to 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,
@@ -97,6 +97,39 @@ public class TotalRequestsThrottlerCleanTest extends
ContextTestSupport {
assertEquals(List.of(TimeUnit.HOURS.toMillis(20)), executor.delays);
}
+ @Test
+ public void testCleanStaysScheduledWhileFarFromFiring() throws Exception {
+ template.sendBody("direct:start", "first");
+ template.sendBody("direct:start", "second");
+
+ // scheduling and cancelling per exchange would take the executor
queue lock about three times
+ assertEquals(1, executor.delays.size());
+ }
+
+ @Test
+ public void testCleanIsRescheduledWhenCloseToFiring() throws Exception {
+ // the clean removes a state whose permits are all returned, even if
they are still delayed, so it must not run
+ // within the period of the permit returned now, or the next exchanges
would get a fresh set of permits
+ executor.pendingDelayMillis = TimeUnit.MINUTES.toMillis(90);
+
+ template.sendBody("direct:start", "first");
+ template.sendBody("direct:start", "second");
+
+ assertEquals(2, executor.delays.size());
+ }
+
+ @Test
+ public void testCleanIsRescheduledWhenTimePeriodIsRaised() throws
Exception {
+ TotalRequestsThrottler throttler = context.getProcessor("throttler",
TotalRequestsThrottler.class);
+
+ template.sendBody("direct:start", "first");
+ // such as from JMX: the pending clean is now within two periods
+ throttler.setTimePeriodMillis(TimeUnit.DAYS.toMillis(1));
+ template.sendBody("direct:start", "second");
+
+ assertEquals(List.of(TimeUnit.HOURS.toMillis(10),
TimeUnit.DAYS.toMillis(10)), executor.delays);
+ }
+
@Override
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
@@ -125,6 +158,7 @@ public class TotalRequestsThrottlerCleanTest extends
ContextTestSupport {
private static final class CapturingExecutor extends
ScheduledThreadPoolExecutor {
private final List<Runnable> scheduled = new CopyOnWriteArrayList<>();
private final List<Long> delays = new CopyOnWriteArrayList<>();
+ private volatile long pendingDelayMillis = TimeUnit.DAYS.toMillis(1);
CapturingExecutor() {
super(1);
@@ -135,7 +169,7 @@ public class TotalRequestsThrottlerCleanTest extends
ContextTestSupport {
public ScheduledFuture<?> schedule(Runnable command, long delay,
TimeUnit unit) {
scheduled.add(command);
delays.add(unit.toMillis(delay));
- return super.schedule(command, 1, TimeUnit.DAYS);
+ return super.schedule(command, pendingDelayMillis,
TimeUnit.MILLISECONDS);
}
void runScheduled() {
