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 861708449a17 CAMEL-26557: camel-util - BoundedExecutorService: a task
of the executor runs nested work inline with the Abort policy too (#29129)
861708449a17 is described below
commit 861708449a1789ee8a312a8e5304638f2d10d374
Author: Adriano Machado <[email protected]>
AuthorDate: Wed Nov 4 03:13:36 2026 -0500
CAMEL-26557: camel-util - BoundedExecutorService: a task of the executor
runs nested work inline with the Abort policy too (#29129)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../impl/DefaultExecutorServiceManagerTest.java | 26 ++++++++-
.../SplitterParallelAsyncProcessorIssueTest.java | 19 +++---
.../camel/processor/WireTapAbortPolicyTest.java | 2 +-
.../management/ManagedRouteRemoveWireTapTest.java | 3 -
.../util/concurrent/BoundedExecutorService.java | 11 ++--
.../concurrent/BoundedExecutorServiceTest.java | 67 ++++++++++++++--------
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +--
.../modules/ROOT/pages/threading-model.adoc | 11 ++--
8 files changed, 93 insertions(+), 55 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExecutorServiceManagerTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExecutorServiceManagerTest.java
index 7f0bd2783883..71209504e21f 100644
---
a/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExecutorServiceManagerTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/DefaultExecutorServiceManagerTest.java
@@ -42,8 +42,6 @@ import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches =
"true",
- disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public class DefaultExecutorServiceManagerTest extends ContextTestSupport {
@Test
@@ -145,6 +143,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testDefaultThreadPool() {
ExecutorService myPool =
context.getExecutorServiceManager().newDefaultThreadPool(this, "myPool");
assertFalse(myPool.isShutdown());
@@ -161,6 +161,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testDefaultUnboundedQueueThreadPool() {
ThreadPoolProfile custom = new ThreadPoolProfile("custom");
custom.setPoolSize(10);
@@ -186,6 +188,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testDefaultNoMaxQueueThreadPool() {
ThreadPoolProfile custom = new ThreadPoolProfile("custom");
custom.setPoolSize(10);
@@ -211,6 +215,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testCustomDefaultThreadPool() {
ThreadPoolProfile custom = new ThreadPoolProfile("custom");
custom.setKeepAliveTime(20L);
@@ -279,6 +285,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testGetThreadPoolProfileInheritDefaultValues() {
assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));
ThreadPoolProfile foo = new ThreadPoolProfile("foo");
@@ -296,6 +304,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testGetThreadPoolProfileInheritCustomDefaultValues() {
ThreadPoolProfile newDefault = new ThreadPoolProfile("newDefault");
newDefault.setKeepAliveTime(30L);
@@ -323,6 +333,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testGetThreadPoolProfileInheritCustomDefaultValues2() {
ThreadPoolProfile newDefault = new ThreadPoolProfile("newDefault");
// just change the max pool as the default profile should then inherit
@@ -347,6 +359,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testNewThreadPoolProfile() {
assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));
@@ -371,6 +385,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testNewThreadPoolProfileById() {
assertNull(context.getExecutorServiceManager().getThreadPoolProfile("foo"));
@@ -397,6 +413,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testNewThreadPoolMinMax() {
ExecutorService pool =
context.getExecutorServiceManager().newThreadPool(this, "Cool", 5, 10);
assertNotNull(pool);
@@ -413,6 +431,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testNewFixedThreadPool() {
ExecutorService pool =
context.getExecutorServiceManager().newFixedThreadPool(this, "Cool", 5);
assertNotNull(pool);
@@ -481,6 +501,8 @@ public class DefaultExecutorServiceManagerTest extends
ContextTestSupport {
}
@Test
+ @DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches
= "true",
+ disabledReason = "In case of Virtual Threads,
ThreadPerTaskExecutor is created instead of ThreadPoolExecutor")
public void testNewCachedThreadPool() {
ExecutorService pool =
context.getExecutorServiceManager().newCachedThreadPool(this, "Cool");
assertNotNull(pool);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelAsyncProcessorIssueTest.java
b/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelAsyncProcessorIssueTest.java
index ce1682224680..570a65ebf1ba 100644
---
a/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelAsyncProcessorIssueTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/issues/SplitterParallelAsyncProcessorIssueTest.java
@@ -16,9 +16,8 @@
*/
package org.apache.camel.issues;
-import java.util.HashSet;
-import java.util.Set;
import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.*;
import org.apache.camel.builder.RouteBuilder;
@@ -30,7 +29,8 @@ import org.junit.jupiter.api.Test;
public class SplitterParallelAsyncProcessorIssueTest extends
ContextTestSupport {
- private final Set<String> threads = new HashSet<>();
+ private final AtomicInteger inFlight = new AtomicInteger();
+ private final AtomicInteger peakInFlight = new AtomicInteger();
@Test
public void testSplitParallelProcessingMaxThreads() throws Exception {
@@ -56,10 +56,10 @@ public class SplitterParallelAsyncProcessorIssueTest
extends ContextTestSupport
// to avoid flaky failures on slow CI
MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
- log.info("{} Threads in use: {}", threads.size(), threads);
-
- // 2 from split EIP and 2 from the async processor that uses the JDK
ForJoinPool
- Assertions.assertTrue(threads.size() <= 4, "Should not use more than 4
threads, was: " + threads.size());
+ // the thread pool of the split allows 2 items at a time. More means
that the async processor let the split
+ // start the next items before the previous ones were done, and fewer
that a permit was lost (in virtual
+ // thread mode, every task gets a new thread, so counting thread names
does not measure this)
+ Assertions.assertEquals(2, peakInFlight.get(), "Items processed at the
same time");
}
@Override
@@ -86,7 +86,6 @@ public class SplitterParallelAsyncProcessorIssueTest extends
ContextTestSupport
public boolean process(Exchange exchange, AsyncCallback
callback) {
// Run the processing in a separate thread
CompletableFuture.runAsync(() -> {
- threads.add(Thread.currentThread().getName());
try {
// Simulate some asynchronous processing
Thread.sleep(250);
@@ -113,10 +112,10 @@ public class SplitterParallelAsyncProcessorIssueTest
extends ContextTestSupport
.split()
.xpath("/employees/employee")
.parallelProcessing().stopOnException().timeout(10000).executorService("threadPoolProfile").synchronous()
- .process(e ->
threads.add(Thread.currentThread().getName()))
+ .process(e ->
peakInFlight.accumulateAndGet(inFlight.incrementAndGet(), Math::max))
.process(asyncProcessor)
- .process(e ->
threads.add(Thread.currentThread().getName()))
.delay(250)
+ .process(e -> inFlight.decrementAndGet())
.end()
.to("mock:split");
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/WireTapAbortPolicyTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/WireTapAbortPolicyTest.java
index d96527a0f8d9..5b279ab7ee47 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/WireTapAbortPolicyTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/WireTapAbortPolicyTest.java
@@ -39,7 +39,7 @@ import static org.junit.jupiter.api.Assertions.fail;
* Wire tap unit test
*/
@DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches =
"true",
- disabledReason = "Tasks are not rejected when using
Virtual Threads")
+ disabledReason = "With Virtual Threads, maxQueueSize
0 means no limit, so no task is rejected")
@Isolated
public class WireTapAbortPolicyTest extends ContextTestSupport {
diff --git
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapTest.java
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapTest.java
index 5e5eead12460..99fe0f054bed 100644
---
a/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapTest.java
+++
b/core/camel-management/src/test/java/org/apache/camel/management/ManagedRouteRemoveWireTapTest.java
@@ -24,7 +24,6 @@ import javax.management.ObjectName;
import org.apache.camel.ServiceStatus;
import org.apache.camel.builder.RouteBuilder;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
@@ -33,8 +32,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@DisabledIfSystemProperty(named = "camel.threads.virtual.enabled", matches =
"true",
- disabledReason = "In case of Virtual Threads, the
thread pools are not ThreadPoolExecutor")
@DisabledOnOs(OS.AIX)
public class ManagedRouteRemoveWireTapTest extends ManagementTestSupport {
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 73233e97e03a..fcd3283f4b79 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
@@ -45,9 +45,10 @@ import java.util.concurrent.atomic.LongAdder;
* </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.
+ * it would wait for. So when no permit is available it runs the work itself
at once, whatever the policy, or in a new
+ * task when that gets too deep, which is also outside the concurrency limit.
Waiting would only delay the work while
+ * the task holds a permit, so a parallel split would run one item fewer at a
time, or, with {@code Abort} and a single
+ * permit, be rejected after the timeout. The policies only apply to a
submitter that is not a task of this executor.
* <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
@@ -97,8 +98,8 @@ 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;
+ // a task of this executor can hold the permit it would wait for, so
it must not wait, whatever the policy
+ Integer depth = inlineDepth.get();
boolean runInline = depth != null;
boolean acquired = false;
try {
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 ffe6b8c5f47b..1660b043b14f 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
@@ -254,6 +254,48 @@ public class BoundedExecutorServiceTest {
assertNestedTaskRunsInline(ThreadPoolRejectedPolicy.Block);
}
+ @Test
+ public void testNestedTaskRunsInlineWithAbortPolicy() throws Exception {
+ // the wait cannot succeed, so the task was rejected after the timeout
and a split with a single permit failed
+ assertNestedTaskRunsInline(ThreadPoolRejectedPolicy.Abort);
+ }
+
+ @Test
+ public void
testNestedTaskRunsInlineWithAbortPolicyWhenOtherTasksHoldThePermits() throws
Exception {
+ // a parallel split holds a permit while it submits its items, so when
it waited for a permit it ran one item
+ // fewer at a time than the executor allows
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(delegate, 2, 60,
TimeUnit.SECONDS, false, ThreadPoolRejectedPolicy.Abort);
+ try {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean ranInline = new AtomicBoolean();
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ // the first item takes the other permit and keeps it
+ sized.execute(() -> {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ // the second item runs while the first one is still running
+ sized.execute(() -> ranInline.set(Thread.currentThread() ==
outer));
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS), "The second item
waited for the first one to finish");
+ assertTrue(ranInline.get(), "The second item should run in the
task that submitted it");
+ assertEquals(1, sized.getCallerRunsCount());
+ assertEquals(0, sized.getRejectedCount());
+ release.countDown();
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
private void assertNestedTaskRunsInline(ThreadPoolRejectedPolicy policy)
throws Exception {
var delegate = Executors.newCachedThreadPool();
var sized = new BoundedExecutorService(delegate, 1, 60,
TimeUnit.SECONDS, false, policy);
@@ -441,31 +483,6 @@ public class BoundedExecutorServiceTest {
}
}
- @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
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 4520961b6bcf..dda0dccee5f1 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -156,10 +156,11 @@ split when it had them. `SplitResult.getProcessedItems()`
now reports the items
=== 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.
+parallel split, no longer waits for a permit when none is available. It runs
the work itself at once, whatever the
+rejected policy. 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. With `Abort`, a
parallel split ran one item fewer at a time
+than `maxQueueSize` allows, and with `maxQueueSize` 1 it was rejected after
`keepAliveTime`. A submitter outside the
+pool waits, and is rejected with `Abort`, as before.
=== camel-core - Routing Slip and Dynamic Router fire the sent event of a step
before the next step
diff --git a/docs/user-manual/modules/ROOT/pages/threading-model.adoc
b/docs/user-manual/modules/ROOT/pages/threading-model.adoc
index 445b202f8d6a..5ad100c4ed3e 100644
--- a/docs/user-manual/modules/ROOT/pages/threading-model.adoc
+++ b/docs/user-manual/modules/ROOT/pages/threading-model.adoc
@@ -329,7 +329,7 @@ message broker consumers and batch workloads where losing a
task is unacceptable
|===
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>>).
+free, whatever the policy, 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>>).
@@ -370,10 +370,11 @@ 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.
+can hold the permit it would wait for. So when no permit is free, that work
runs at once on the submitting thread
+instead of waiting, whatever the `rejectedPolicy`, also when other tasks hold
the permits. Waiting would only delay the
+work while the task holds a permit: a parallel `split` would run one item
fewer at a time than the limit, or wait for
+the permit it holds itself. Such work runs outside the concurrency limit, as
does the work handed to a new task when it
+is nested too deeply. The `rejectedPolicy` only applies to a submitter that is
not a task of the pool.
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
