This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 59da736022ae [backport camel-4.22.x] CAMEL-26478: camel-util -
BoundedExecutorService: a task must not wait for a permit it holds itself
(#28996)
59da736022ae is described below
commit 59da736022aee785ad9e5ea0063b52a13f386a3a
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Nov 4 09:20:52 2026 +0100
[backport camel-4.22.x] CAMEL-26478: camel-util - BoundedExecutorService: a
task must not wait for a permit it holds itself (#28996)
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
---
.../util/concurrent/BoundedExecutorService.java | 63 ++++-
.../concurrent/BoundedExecutorServiceTest.java | 261 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 8 +
.../modules/ROOT/pages/threading-model.adoc | 9 +
4 files changed, 337 insertions(+), 4 deletions(-)
diff --git
a/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
b/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
index 6111377dcce5..73233e97e03a 100644
---
a/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
+++
b/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
@@ -44,6 +44,11 @@ import java.util.concurrent.atomic.LongAdder;
* <li><b>Block</b>: blocks indefinitely until a permit becomes available. No
timeout, no rejection.</li>
* </ul>
* <p>
+ * A task of this executor that submits more work, such as a parallel split
that submits its items, can hold the permit
+ * it would wait for. So when no permit is available it runs the work itself
at once, with the {@code CallerRuns} and
+ * {@code Block} policies, or in a new task when that gets too deep, which is
also outside the concurrency limit. With
+ * {@code Abort} it waits and is rejected as usual.
+ * <p>
* <b>Caller thread blocking:</b> while waiting for a permit, the calling
thread is blocked. When callers are virtual
* threads this is inexpensive (the carrier thread is released). When callers
are platform threads (e.g., HTTP server
* threads) the blocked thread is unavailable for other work — this is
standard backpressure behavior but worth noting
@@ -52,11 +57,17 @@ import java.util.concurrent.atomic.LongAdder;
*/
public class BoundedExecutorService extends AbstractExecutorService {
+ // Bounds the stack of tasks running inline. Past it, the work goes to a
new task of the delegate. That is the escape
+ // hatch against the deadlock, accepted at the cost of the concurrency
limit, as the inline runs are also outside it.
+ private static final int MAX_INLINE_DEPTH = 64;
+
private final ExecutorService delegate;
private final Semaphore semaphore;
private final int maxConcurrent;
private final long timeoutNanos;
private final ThreadPoolRejectedPolicy rejectedPolicy;
+ // how deep tasks are running inline in a task of this executor, and null
in any other thread
+ private final ThreadLocal<Integer> inlineDepth = new ThreadLocal<>();
private final LongAdder callerRunsCount = new LongAdder();
private final LongAdder rejectedCount = new LongAdder();
private final LongAdder delegatedTaskCount = new LongAdder();
@@ -86,9 +97,14 @@ public class BoundedExecutorService extends
AbstractExecutorService {
throw new RejectedExecutionException("Executor has been shut
down");
}
+ // a task of this executor can hold the permit it would wait for, so
it must not wait
+ Integer depth = rejectedPolicy != ThreadPoolRejectedPolicy.Abort ?
inlineDepth.get() : null;
+ boolean runInline = depth != null;
boolean acquired = false;
try {
- if (rejectedPolicy == ThreadPoolRejectedPolicy.Block) {
+ if (runInline) {
+ acquired = semaphore.tryAcquire();
+ } else if (rejectedPolicy == ThreadPoolRejectedPolicy.Block) {
semaphore.acquire();
acquired = true;
} else {
@@ -96,6 +112,10 @@ public class BoundedExecutorService extends
AbstractExecutorService {
}
if (!acquired) {
+ if (runInline) {
+ runNested(command, depth);
+ return;
+ }
if (rejectedPolicy == ThreadPoolRejectedPolicy.CallerRuns) {
callerRunsCount.increment();
command.run();
@@ -109,9 +129,8 @@ public class BoundedExecutorService extends
AbstractExecutorService {
try {
delegate.execute(() -> {
try {
- command.run();
+ runTask(command);
} finally {
- delegatedTaskCount.increment();
semaphore.release();
}
});
@@ -127,6 +146,41 @@ public class BoundedExecutorService extends
AbstractExecutorService {
}
}
+ private void runTask(Runnable command) {
+ // a delegate can run the task on the calling thread, which can be a
task of this executor itself
+ Integer previous = inlineDepth.get();
+ inlineDepth.set(0);
+ try {
+ command.run();
+ } finally {
+ if (previous == null) {
+ inlineDepth.remove();
+ } else {
+ inlineDepth.set(previous);
+ }
+ delegatedTaskCount.increment();
+ }
+ }
+
+ /**
+ * Runs a task submitted by a task of this executor when no permit is
available. It must not wait, as it can hold
+ * that permit itself, and it must not wait for a queue to run later
either, as it could be waiting for this task.
+ */
+ private void runNested(Runnable command, int depth) {
+ if (depth >= MAX_INLINE_DEPTH) {
+ // a task that submits itself again, as a parallel split does for
each item, must not overflow the stack
+ delegate.execute(() -> runTask(command));
+ return;
+ }
+ callerRunsCount.increment();
+ inlineDepth.set(depth + 1);
+ try {
+ command.run();
+ } finally {
+ inlineDepth.set(depth);
+ }
+ }
+
// -- Metrics --
/**
@@ -159,7 +213,8 @@ public class BoundedExecutorService extends
AbstractExecutorService {
}
/**
- * The number of times the timeout expired and a task fell back to running
on the caller's thread.
+ * The number of times a task ran on the caller's thread, because the
timeout expired or a task of this executor
+ * submitted it when no permit was available and the inline depth limit
was not reached.
*/
public long getCallerRunsCount() {
return callerRunsCount.sum();
diff --git
a/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
b/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
index 1b58012f35d9..ffe6b8c5f47b 100644
---
a/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
+++
b/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.util.concurrent;
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -29,6 +31,7 @@ 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.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -238,4 +241,262 @@ public class BoundedExecutorServiceTest {
delegate.shutdown();
}
}
+
+ @Test
+ public void testNestedTaskRunsInlineWhenItHoldsTheOnlyPermit() throws
Exception {
+ // a parallel split runs in the executor and submits its items to it,
so waiting cannot succeed
+ assertNestedTaskRunsInline(ThreadPoolRejectedPolicy.CallerRuns);
+ }
+
+ @Test
+ public void testNestedTaskRunsInlineWithBlockPolicy() throws Exception {
+ // this wait has no timeout, so the task would hang forever
+ assertNestedTaskRunsInline(ThreadPoolRejectedPolicy.Block);
+ }
+
+ private void assertNestedTaskRunsInline(ThreadPoolRejectedPolicy policy)
throws Exception {
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(delegate, 1, 60,
TimeUnit.SECONDS, false, policy);
+ try {
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean sameThread = new AtomicBoolean();
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ sized.execute(() -> {
+ sameThread.set(Thread.currentThread() == outer);
+ done.countDown();
+ });
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS), "The nested task
waited for the permit its parent holds");
+ assertTrue(sameThread.get(), "The nested task should run in the
task that submitted it");
+ assertEquals(1, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testTaskThatResubmitsItselfDoesNotOverflowTheStack() throws
Exception {
+ // a parallel split submits itself again for each item, so a large
split must not run all of them inline
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 1, 60, TimeUnit.SECONDS, false,
ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ int resubmits = 200_000;
+ AtomicInteger count = new AtomicInteger();
+ CountDownLatch done = new CountDownLatch(1);
+ Runnable[] task = new Runnable[1];
+ task[0] = () -> {
+ if (count.incrementAndGet() == resubmits) {
+ done.countDown();
+ } else {
+ sized.execute(task[0]);
+ }
+ };
+
+ sized.execute(task[0]);
+
+ assertTrue(done.await(30, TimeUnit.SECONDS), "Stopped after " +
count.get() + " resubmits");
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testNestedTaskThatThrowsDoesNotUseUpTheInlineDepth() throws
Exception {
+ // the depth must be restored on failure too, or a task that survives
failures of the work it submits would
+ // reach the depth limit and hand the work to new tasks, outside the
concurrency limit
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 1, 60, TimeUnit.SECONDS, false,
ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ int failures = 100;
+ AtomicInteger inline = new AtomicInteger();
+ CountDownLatch done = new CountDownLatch(1);
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ for (int i = 0; i < failures; i++) {
+ try {
+ sized.execute(() -> {
+ if (Thread.currentThread() == outer) {
+ inline.incrementAndGet();
+ }
+ throw new IllegalStateException("Forced");
+ });
+ } catch (IllegalStateException e) {
+ // an inline task fails in its caller, as a caller-run
task does
+ }
+ }
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertEquals(failures, inline.get(), "Every nested task should
still run inline after a failure");
+ assertEquals(failures, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void
testDelegateRunningATaskOnTheCallingThreadKeepsTheDepthOfTheOuterTask() throws
Exception {
+ // a delegate that runs a task on the calling thread must not reset
the depth of the task that submitted it,
+ // or that task would wait for a permit it holds itself
+ var pool = Executors.newCachedThreadPool();
+ ThreadLocal<Boolean> direct = new ThreadLocal<>();
+ var delegate = new AbstractExecutorService() {
+ @Override
+ public void execute(Runnable command) {
+ if (direct.get() != null) {
+ command.run();
+ } else {
+ pool.execute(command);
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ pool.shutdown();
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return pool.shutdownNow();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return pool.isShutdown();
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return pool.isTerminated();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
+ return pool.awaitTermination(timeout, unit);
+ }
+ };
+ var sized = new BoundedExecutorService(delegate, 2, 60,
TimeUnit.SECONDS, false, ThreadPoolRejectedPolicy.Block);
+ try {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean ranInline = new AtomicBoolean();
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ // a free permit: the delegate runs this one on the calling
thread
+ direct.set(true);
+ sized.execute(() -> {
+ });
+ direct.remove();
+ // take the other permit, so that the next task finds none
+ sized.execute(() -> {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ // with Block this waits forever if the depth of this task was
lost
+ sized.execute(() -> ranInline.set(Thread.currentThread() ==
outer));
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS), "The nested task
should not wait for the permit");
+ assertTrue(ranInline.get(), "The nested task should run inline");
+ release.countDown();
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testNestedTaskUsesAFreePermit() throws Exception {
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 2, 60, TimeUnit.SECONDS, false,
ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean sameThread = new AtomicBoolean(true);
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ sized.execute(() -> {
+ sameThread.set(Thread.currentThread() == outer);
+ done.countDown();
+ });
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertFalse(sameThread.get(), "The bound still allows this task to
run concurrently");
+ assertEquals(0, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testNestedTaskIsStillRejectedWithAbortPolicy() throws
Exception {
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 1, 100, TimeUnit.MILLISECONDS, false,
ThreadPoolRejectedPolicy.Abort);
+ try {
+ CountDownLatch rejected = new CountDownLatch(1);
+
+ sized.execute(() -> {
+ try {
+ sized.execute(() -> {
+ // never runs
+ });
+ } catch (RejectedExecutionException e) {
+ rejected.countDown();
+ }
+ });
+
+ assertTrue(rejected.await(5, TimeUnit.SECONDS), "Abort must fail
the task and not run it");
+ assertEquals(0, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testTaskOfAnotherExecutorStillWaits() throws Exception {
+ // only a task of this executor can hold one of its permits, so others
wait for backpressure as usual
+ var full = new BoundedExecutorService(
+ Executors.newCachedThreadPool(), 1, 200,
TimeUnit.MILLISECONDS, false, ThreadPoolRejectedPolicy.CallerRuns);
+ var other = new BoundedExecutorService(
+ Executors.newCachedThreadPool(), 1, 60, TimeUnit.SECONDS,
false, ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicInteger waitedMillis = new AtomicInteger();
+ full.execute(() -> {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ other.execute(() -> {
+ long start = System.nanoTime();
+ full.execute(() -> waitedMillis.set((int)
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)));
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertTrue(waitedMillis.get() >= 150, "It should wait for the
timeout, but waited " + waitedMillis.get());
+ release.countDown();
+ } finally {
+ full.shutdownNow();
+ other.shutdownNow();
+ }
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 780d9f5fdb33..5f6313200ede 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -2296,6 +2296,14 @@ CamelContext of that provider. It now closes each of
them only when no other `JC
it returned the closed cache, whose operations throw `IllegalStateException`.
This matters for code that calls a
`JCacheManager` (for example one created by a custom `JCacheManagerFactory`)
after it was closed.
+=== camel-util - BoundedExecutorService: a task of the executor does not wait
for a permit
+
+With virtual threads, a task that runs in a thread pool bounded by
`maxQueueSize` and submits more work to it, such as a
+parallel split, no longer waits for a permit when none is available. It runs
the work itself at once, with the
+`CallerRuns` and `Block` rejected policies. Before, it could wait for a permit
it held itself, until `keepAliveTime`
+expired, or forever with `Block`. A split with a `timeout` could lose items
that way. A submitter outside the pool waits
+as before.
+
=== camel-telemetry, camel-tracing - the http.url, url.path and server.address
span attributes are presented like camel.uri
This applies to `camel-opentelemetry2`, `camel-micrometer-observability`,
`camel-telemetry-dev`, `camel-opentelemetry`
diff --git a/docs/user-manual/modules/ROOT/pages/threading-model.adoc
b/docs/user-manual/modules/ROOT/pages/threading-model.adoc
index b419a3a1410a..8b2ea9065d3a 100644
--- a/docs/user-manual/modules/ROOT/pages/threading-model.adoc
+++ b/docs/user-manual/modules/ROOT/pages/threading-model.adoc
@@ -327,6 +327,9 @@ systems where failing fast is preferred over blocking.
message broker consumers and batch workloads where losing a task is
unacceptable and latency is less critical.
|===
+With virtual threads, a task of the pool that submits more work to the same
pool runs it at once when no permit is
+free, for `CallerRuns` and `Block`, instead of waiting (see
<<virtual-threads>>).
+
With platform threads, these policies apply when the `ThreadPoolExecutor` work
queue is full. With virtual threads,
the same policies apply when the concurrency semaphore has no available
permits (see <<virtual-threads>>).
@@ -363,6 +366,12 @@ the task runs on the caller's thread.
a `RejectedExecutionException` is thrown.
* *Block*: the caller blocks indefinitely until a permit becomes available.
+A task of the pool that submits more work to the same pool (for example a
parallel `split` that submits its items)
+can hold the permit it would wait for. So with `CallerRuns` and `Block`, when
no permit is free, that work runs at once
+on the submitting thread instead of waiting. For `Block` this applies even
when other tasks hold the permits. Such work
+runs outside the concurrency limit, as does the work handed to a new task when
it is nested too deeply. `Abort` is
+not affected: it still waits and is rejected as usual.
+
While waiting for a permit, the calling thread is blocked. When callers are
virtual threads this is inexpensive
(the carrier thread is released). When callers are platform threads (e.g.,
HTTP server threads) the blocked thread
is unavailable for other work.
