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 02d2ce6793b6 CAMEL-26309: camel-core - Aggregate EIP: fail the
aggregated exchange when the thread pool rejects it (#28709)
02d2ce6793b6 is described below
commit 02d2ce6793b69e50e4112fc114c284c74ecec11f
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 30 14:22:38 2026 +0530
CAMEL-26309: camel-core - Aggregate EIP: fail the aggregated exchange when
the thread pool rejects it (#28709)
… when the thread pool rejects it
When the thread pool of the aggregator rejected the task that sends an
aggregated exchange, the group had already been removed from the
aggregation repository and booked as in progress, but nothing ran the
on completions of the aggregated exchange. The group was lost: a
recoverable repository never recovered it (its id stayed in progress),
the memory repository dropped it without a trace, the in progress count
never went back to 0 (forceCompletionOnStop then waits until the
shutdown timeout), and the on completions of the aggregation strategy
and of spooled stream caches never ran. When the batch consumer
completed several groups at once, the groups after the rejected one
were not even submitted, and were lost the same way.
Now the aggregated exchange fails with the RejectedExecutionException:
its on completions run (AggregateOnCompletion.onFailure undoes the
booking and does not confirm it, so a recoverable repository recovers
it), it is passed to the exception handler, and the caller still fails
with the rejection as before (AggregateParallelThreadPoolAbortTest).
The other groups of a batch are still sent, or failed the same way,
before the first rejection is thrown.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/aggregate/AggregateProcessor.java | 91 +++++--
.../AggregateRejectedCompletionTest.java | 272 +++++++++++++++++++++
2 files changed, 338 insertions(+), 25 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
index f7cf84d3fcf9..087991e9681f 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
@@ -25,6 +25,7 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -507,7 +508,7 @@ public class AggregateProcessor extends BaseProcessorSupport
// we are completed so submit to completion outside the lock. This
must also be done when the aggregation
// failed, or must be retried due to optimistic locking, after a
group was completed (such as a group
// completed by pre-completion), as that group has already been
removed from the repository
- aggregated.forEach(agg -> onSubmitCompletion(key, agg));
+ submitCompletions(key, aggregated);
}
// check for the special header to force completion of all groups
(inclusive of the message)
@@ -1095,6 +1096,27 @@ public class AggregateProcessor extends
BaseProcessorSupport
return false;
}
+ /**
+ * Sends the aggregated exchanges (more than one when the batch consumer
completes all its groups). None of them is
+ * in the repository any more, so when the thread pool rejects one of
them, the others are still sent (or failed if
+ * rejected too), and then the first rejection is thrown.
+ */
+ private void submitCompletions(String key, List<Exchange> aggregated) {
+ RejectedExecutionException rejected = null;
+ for (Exchange agg : aggregated) {
+ try {
+ onSubmitCompletion(key, agg);
+ } catch (RejectedExecutionException e) {
+ if (rejected == null) {
+ rejected = e;
+ }
+ }
+ }
+ if (rejected != null) {
+ throw rejected;
+ }
+ }
+
private void onSubmitCompletion(final String key, final Exchange exchange)
{
LOG.debug("Aggregation complete for correlation key {} sending
aggregated exchange: {}", key, exchange);
@@ -1121,33 +1143,52 @@ public class AggregateProcessor extends
BaseProcessorSupport
exchange.getExchangeExtension().addOnCompletion(new
AggregateOnCompletion(exchange.getExchangeId()));
// send this exchange
- executorService.execute(() -> {
- ExchangeHelper.prepareOutToIn(exchange);
-
- AsyncCallback completionCallback = done -> {
- // log exception if there was a problem
- if (exchange.getException() != null) {
- // if there was an exception then let the exception
handler handle it
- getExceptionHandler().handleException("Error processing
aggregated exchange", exchange,
- exchange.getException());
- } else {
- LOG.trace("Processing aggregated exchange: {} complete.",
exchange);
- }
- };
+ try {
+ executorService.execute(() -> {
+ ExchangeHelper.prepareOutToIn(exchange);
+
+ AsyncCallback completionCallback = done -> {
+ // log exception if there was a problem
+ if (exchange.getException() != null) {
+ // if there was an exception then let the exception
handler handle it
+ getExceptionHandler().handleException("Error
processing aggregated exchange", exchange,
+ exchange.getException());
+ } else {
+ LOG.trace("Processing aggregated exchange: {}
complete.", exchange);
+ }
+ };
- if (executorService instanceof SynchronousExecutorService) {
- // CAMEL-23281: process inline to avoid deadlock with
transacted exchanges
- processor.process(exchange, completionCallback);
- } else {
- Runnable task = () -> processor.process(exchange,
completionCallback);
- // execute the task using this thread sync (similar to
multicast eip in parallel mode)
- if (exchange.isTransacted()) {
- reactiveExecutor.scheduleQueue(task);
+ if (executorService instanceof SynchronousExecutorService) {
+ // CAMEL-23281: process inline to avoid deadlock with
transacted exchanges
+ processor.process(exchange, completionCallback);
} else {
- reactiveExecutor.scheduleSync(task);
+ Runnable task = () -> processor.process(exchange,
completionCallback);
+ // execute the task using this thread sync (similar to
multicast eip in parallel mode)
+ if (exchange.isTransacted()) {
+ reactiveExecutor.scheduleQueue(task);
+ } else {
+ reactiveExecutor.scheduleSync(task);
+ }
}
- }
- });
+ });
+ } catch (RejectedExecutionException e) {
+ onRejectedCompletion(exchange, e);
+ // and let the caller fail as before (such as the exchange that
completed the group)
+ throw e;
+ }
+ }
+
+ /**
+ * The thread pool rejected the task that sends the aggregated exchange.
The group has already been removed from the
+ * aggregation repository, so the aggregated exchange fails as if its
processing had failed: its on completions run
+ * (the aggregator then no longer counts it as in progress, and does not
confirm it, so a recoverable repository
+ * recovers it), and it is passed to the exception handler.
+ */
+ private void onRejectedCompletion(Exchange exchange,
RejectedExecutionException e) {
+ exchange.setException(e);
+ UnitOfWorkHelper.doneSynchronizations(exchange,
exchange.getExchangeExtension().handoverCompletions());
+ getExceptionHandler().handleException("Error processing aggregated
exchange: the thread pool rejected it",
+ exchange, e);
}
private void aggregateCompletionCounter(Exchange exchange) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateRejectedCompletionTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateRejectedCompletionTest.java
new file mode 100644
index 000000000000..78c98ea5f7a7
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateRejectedCompletionTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.aggregator;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.BodyInAggregatingStrategy;
+import org.apache.camel.processor.aggregate.AggregateProcessor;
+import org.apache.camel.spi.AggregationRepository;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.spi.Synchronization;
+import org.apache.camel.spi.ThreadPoolProfile;
+import org.apache.camel.support.KeyValueAggregationRepository;
+import org.apache.camel.support.MemoryKeyValueRepository;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy;
+import org.junit.jupiter.api.AfterEach;
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * When the thread pool of the aggregator rejects the task that sends an
aggregated exchange, the group has already been
+ * removed from the aggregation repository. The aggregated exchange must then
be failed, so that its on completions run,
+ * the aggregator no longer counts it as in progress, and a recoverable
repository recovers it. When the batch consumer
+ * completes several groups at once, this holds for each of them. The exchange
that completed the group still fails with
+ * the rejection, as AggregateParallelThreadPoolAbortTest expects.
+ */
+class AggregateRejectedCompletionTest extends ContextTestSupport {
+
+ private final CountDownLatch blocked = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final List<String> failures = new CopyOnWriteArrayList<>();
+ private final List<Exchange> handled = new CopyOnWriteArrayList<>();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Override
+ @AfterEach
+ public void tearDown() throws Exception {
+ release.countDown();
+ super.tearDown();
+ }
+
+ @Test
+ void testRejectedAggregatedExchangeIsFailed() throws Exception {
+ AggregateProcessor aggregator = addRoute(null, false);
+
+ getMockEndpoint("mock:aggregated").expectedBodiesReceived("A1+A2",
"B1+B2");
+
+ fillThreadPool();
+ // the thread pool is busy with A1+A2, and its queue holds B1+B2, so
the task for C1+C2 is rejected
+ sendBody("C1", "C");
+ Exchange last = send("C2", "C");
+
+ release.countDown();
+ assertMockEndpointsSatisfied();
+
+ // the aggregated exchange is handed to the exception handler, and its
on completions run as failed
+ assertEquals(1, handled.size());
+ assertEquals("C1+C2", handled.get(0).getIn().getBody(String.class));
+ assertInstanceOf(RejectedExecutionException.class,
handled.get(0).getException());
+ assertEquals(List.of("C1+C2"), failures);
+ // and it is no longer in progress
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(0,
aggregator.getInProgressCompleteExchanges()));
+ // the message that completed the group fails with the rejection, as
before
+ assertInstanceOf(RejectedExecutionException.class,
last.getException());
+ }
+
+ @Test
+ void testRejectedAggregatedExchangeIsRecovered() throws Exception {
+ addRoute(recoverableRepository(), false);
+
+
getMockEndpoint("mock:aggregated").expectedBodiesReceivedInAnyOrder("A1+A2",
"B1+B2", "C1+C2");
+
+ fillThreadPool();
+ sendBody("C1", "C");
+ send("C2", "C");
+
+ release.countDown();
+ // the rejected C1+C2 is still in the completed store of the
repository, so it is recovered
+ assertMockEndpointsSatisfied();
+ }
+
+ @Test
+ void testRejectedBatchCompletionFailsEveryGroup() throws Exception {
+ AggregateProcessor aggregator = addRoute(null, true);
+
+ getMockEndpoint("mock:aggregated").expectedBodiesReceived("A1", "B1");
+
+ fillThreadPoolByBatches();
+ // the batch consumer completes the groups C, D and E at once, and the
thread pool rejects all three tasks
+ sendBatch("C1", "C");
+ sendBatch("D1", "D");
+ Exchange last = sendBatch("E1", "E");
+
+ release.countDown();
+ assertMockEndpointsSatisfied();
+
+ // every group of the batch is failed, not only the first one that was
rejected
+ assertEquals(List.of("C1", "D1", "E1"), handled.stream().map(e ->
e.getIn().getBody(String.class)).toList());
+ assertEquals(List.of("C1", "D1", "E1"), failures);
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(0,
aggregator.getInProgressCompleteExchanges()));
+ assertInstanceOf(RejectedExecutionException.class,
last.getException());
+ }
+
+ @Test
+ void testRejectedBatchCompletionIsRecovered() throws Exception {
+ addRoute(recoverableRepository(), true);
+
+ // E1 completed its group on arrival, so its group was never stored in
the repository: its sender gets the
+ // rejection instead
+
getMockEndpoint("mock:aggregated").expectedBodiesReceivedInAnyOrder("A1", "B1",
"C1", "D1");
+
+ fillThreadPoolByBatches();
+ sendBatch("C1", "C");
+ sendBatch("D1", "D");
+ sendBatch("E1", "E");
+
+ release.countDown();
+ // the rejected C1 and D1 are still in the completed store of the
repository, so both are recovered
+ assertMockEndpointsSatisfied();
+ }
+
+ private static KeyValueAggregationRepository recoverableRepository() {
+ KeyValueAggregationRepository repository = new
KeyValueAggregationRepository(new MemoryKeyValueRepository());
+ repository.setUseRecovery(true);
+ repository.setRecoveryInterval(100);
+ return repository;
+ }
+
+ private void fillThreadPoolByBatches() throws InterruptedException {
+ // batches of one message each: A blocks the thread and B waits in the
queue
+ sendBatch("A1", "A", 1);
+ assertTrue(blocked.await(5, TimeUnit.SECONDS));
+ sendBatch("B1", "B", 1);
+ }
+
+ private Exchange sendBatch(String body, String id) {
+ return sendBatch(body, id, 3);
+ }
+
+ private Exchange sendBatch(String body, String id, int batchSize) {
+ return template.send("direct:start", exchange -> {
+ exchange.getIn().setBody(body);
+ exchange.getIn().setHeader("id", id);
+ exchange.setProperty(Exchange.BATCH_SIZE, batchSize);
+ });
+ }
+
+ private void fillThreadPool() throws InterruptedException {
+ sendBody("A1", "A");
+ sendBody("A2", "A");
+ assertTrue(blocked.await(5, TimeUnit.SECONDS));
+ sendBody("B1", "B");
+ sendBody("B2", "B");
+ }
+
+ private void sendBody(String body, String id) {
+ template.sendBodyAndHeader("direct:start", body, "id", id);
+ }
+
+ private Exchange send(String body, String id) {
+ return template.send("direct:start", exchange -> {
+ exchange.getIn().setBody(body);
+ exchange.getIn().setHeader("id", id);
+ });
+ }
+
+ private AggregateProcessor addRoute(AggregationRepository repository,
boolean batch) throws Exception {
+ addRoutes(repository, batch);
+ AggregateProcessor aggregator = context.getProcessor("aggregator",
AggregateProcessor.class);
+ ExceptionHandler handler = new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ handleException(null, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handleException(message, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ if (exchange != null) {
+ handled.add(exchange);
+ }
+ }
+ };
+ aggregator.setExceptionHandler(handler);
+ return aggregator;
+ }
+
+ private void addRoutes(AggregationRepository repository, boolean batch)
throws Exception {
+ ThreadPoolProfile profile = new ThreadPoolProfile("oneThread");
+ profile.setPoolSize(1);
+ profile.setMaxPoolSize(1);
+ profile.setMaxQueueSize(1);
+ profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort);
+ context.getExecutorServiceManager().registerThreadPoolProfile(profile);
+
+ // the strategy adds an on completion to the aggregated exchange, as
the zip and tar strategies do to delete
+ // their temporary file
+ BodyInAggregatingStrategy strategy = new BodyInAggregatingStrategy() {
+ @Override
+ public void onCompletion(Exchange exchange) {
+ String body = exchange.getIn().getBody(String.class);
+ Synchronization cleanup = new SynchronizationAdapter() {
+ @Override
+ public void onFailure(Exchange exchange) {
+ failures.add(body);
+ }
+ };
+ exchange.getExchangeExtension().addOnCompletion(cleanup);
+ }
+ };
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ var aggregate = from("direct:start")
+ .aggregate(header("id"), strategy).id("aggregator")
+ .executorService("oneThread");
+ if (batch) {
+ aggregate.completionFromBatchConsumer();
+ } else {
+ aggregate.completionSize(2);
+ }
+ if (repository != null) {
+ aggregate.aggregationRepository(repository);
+ }
+ aggregate.process(exchange -> {
+ if
(exchange.getIn().getBody(String.class).startsWith("A1")) {
+ blocked.countDown();
+ assertTrue(release.await(5, TimeUnit.SECONDS));
+ }
+ }).to("mock:aggregated");
+ }
+ });
+ context.start();
+ }
+}
