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 c949b429a938 CAMEL-24944: camel-core - Aggregate EIP: send a
pre-completed group also when the next exchange fails to aggregate
c949b429a938 is described below
commit c949b429a938a83852011c3f19200cdcb45d1f09
Author: smjain <[email protected]>
AuthorDate: Wed Sep 23 18:07:51 2026 +0530
CAMEL-24944: camel-core - Aggregate EIP: send a pre-completed group also
when the next exchange fails to aggregate
When the AggregationStrategy pre-completes a group (canPreComplete),
doAggregation
removes the old group from the repository and keeps it in a local list, then
aggregates the new exchange and adds it as a new group. If that second part
did
not finish normally the list was dropped and the completed group was never
sent:
- with optimistic locking the add of the new group fails with an
OptimisticLockingException when another exchange created a group for the
same
key in between; the exception left doAggregation (only
CamelExchangeException
is caught) and the exchange was retried from scratch,
- the strategy throws when aggregating the new exchange,
- the same with discardOnAggregationFailure, where doAggregation returned
null.
The group had already been removed from the repository (and the key closed
and its timeout removed), so it was lost, or only re-delivered later by the
recover task of a recoverable repository.
doAggregation now adds completed exchanges to a list owned by doProcess,
which
submits them after releasing the lock also when doAggregation throws or the
exchange is retried.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/aggregate/AggregateProcessor.java | 24 +--
.../AggregatePreCompleteLostGroupTest.java | 213 +++++++++++++++++++++
2 files changed, 223 insertions(+), 14 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 289360076d45..9bb85dc6f065 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
@@ -448,7 +448,7 @@ public class AggregateProcessor extends BaseProcessorSupport
removeFlagCompleteAllGroups(copy);
removeFlagCompleteAllGroupsInclusive(copy);
- List<Exchange> aggregated = null;
+ List<Exchange> aggregated = new ArrayList<>();
lock.lock();
try {
// check again under the lock (and on every optimistic locking
retry), as the key may have been closed
@@ -456,16 +456,14 @@ public class AggregateProcessor extends
BaseProcessorSupport
if (closedCorrelationKeys != null &&
closedCorrelationKeys.containsKey(key)) {
throw new ClosedCorrelationKeyException(key, exchange);
}
- aggregated = doAggregation(key, copy);
+ doAggregation(key, copy, aggregated);
} catch (CamelExchangeException e) {
exchange.setException(e);
} finally {
lock.unlock();
- }
-
- // we are completed so do that work outside the lock
- if (aggregated != null) {
- // we are completed so submit to completion
+ // 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));
}
@@ -516,20 +514,19 @@ public class AggregateProcessor extends
BaseProcessorSupport
* <p/>
* This method <b>must</b> be run synchronized as we cannot aggregate the
same correlation key in parallel.
* <p/>
- * The returned {@link Exchange} should be send downstream using the
+ * The completed {@link Exchange}(s) are added to the given list as soon
as they are completed (and removed from the
+ * repository), also if this method fails afterwards. They should be send
downstream using the
* {@link #onSubmitCompletion(String, org.apache.camel.Exchange)} method
which sends out the aggregated and
* completed {@link Exchange}.
*
* @param key the correlation key
* @param newExchange the exchange
- * @return the aggregated
exchange(s) which is complete, or <tt>null</tt> if
- * not yet complete
+ * @param list the list to add the
aggregated exchange(s) which are complete to
* @throws org.apache.camel.CamelExchangeException is thrown if error
aggregating
*/
- private List<Exchange> doAggregation(String key, Exchange newExchange)
throws CamelExchangeException {
+ private void doAggregation(String key, Exchange newExchange,
List<Exchange> list) throws CamelExchangeException {
LOG.trace("onAggregation +++ start +++ with correlation key: {}", key);
- List<Exchange> list = new ArrayList<>();
String complete = null;
Exchange answer;
@@ -611,7 +608,7 @@ public class AggregateProcessor extends BaseProcessorSupport
answer = oldExchange;
if (answer == null) {
// first message in group failed during aggregation and we
should just discard this
- return null;
+ return;
}
} else {
// must catch any exception from aggregation
@@ -663,7 +660,6 @@ public class AggregateProcessor extends BaseProcessorSupport
}
LOG.trace("onAggregation +++ end +++ with correlation key: {}", key);
- return list;
}
protected void doAggregationComplete(
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregatePreCompleteLostGroupTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregatePreCompleteLostGroupTest.java
new file mode 100644
index 000000000000..aa806fd1f3b1
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregatePreCompleteLostGroupTest.java
@@ -0,0 +1,213 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.SendProcessor;
+import org.apache.camel.processor.aggregate.AggregateProcessor;
+import org.apache.camel.processor.aggregate.MemoryAggregationRepository;
+import org.apache.camel.processor.aggregate.OptimisticLockRetryPolicy;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * A group completed by pre-completion has been removed from the repository,
so it must be sent even if the exchange
+ * that pre-completed it cannot be aggregated afterwards.
+ */
+public class AggregatePreCompleteLostGroupTest extends ContextTestSupport {
+
+ private final CountDownLatch removed = new CountDownLatch(1);
+ private final CountDownLatch releaseRemove = new CountDownLatch(1);
+ private ExecutorService executorService;
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ executorService = Executors.newSingleThreadExecutor();
+ }
+
+ @Override
+ @AfterEach
+ public void tearDown() throws Exception {
+ releaseRemove.countDown();
+ executorService.shutdownNow();
+ super.tearDown();
+ }
+
+ @Test
+ public void testOptimisticLockingFailureAfterPreCompletion() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a1", "c", "START-b");
+
+ // pauses the thread of START-b right after it removed the
pre-completed group from the repository
+ AtomicBoolean pauseRemove = new AtomicBoolean(true);
+ MemoryAggregationRepository repository = new
MemoryAggregationRepository(true) {
+ @Override
+ public void remove(CamelContext camelContext, String key, Exchange
exchange) {
+ super.remove(camelContext, key, exchange);
+ if
(Thread.currentThread().getName().equals("producer-START-b") &&
pauseRemove.getAndSet(false)) {
+ removed.countDown();
+ await(releaseRemove);
+ }
+ }
+ };
+
+ AggregateProcessor ap = createProcessor();
+ ap.setAggregationRepository(repository);
+ ap.setOptimisticLocking(true);
+ // retry at once in the same thread
+ ap.setOptimisticLockRetryPolicy(new
OptimisticLockRetryPolicy().retryDelay(0).maximumRetries(5));
+ ap.start();
+
+ ap.process(createExchange("a1"));
+
+ // START-b pre-completes the group [a1]: it is removed from the
repository, and START-b is paused
+ Exchange b = createExchange("START-b");
+ Thread producer = new Thread(() -> process(ap, b), "producer-START-b");
+ producer.start();
+ await(removed);
+
+ // c starts a new group, so START-b fails to add its new group and is
retried
+ ap.process(createExchange("c"));
+ releaseRemove.countDown();
+ producer.join(10000);
+ assertNull(b.getException());
+
+ // pre-complete the group of START-b
+ ap.process(createExchange("START-end"));
+
+ assertMockEndpointsSatisfied();
+ ap.stop();
+ }
+
+ @Test
+ public void testAggregationFailureAfterPreCompletion() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a1");
+
+ AggregateProcessor ap = createProcessor();
+ ap.start();
+
+ ap.process(createExchange("a1"));
+ Exchange bad = createExchange("START-fail");
+ ap.process(bad);
+
+ assertNotNull(bad.getException());
+ assertTrue(ap.getAggregationRepository().getKeys().isEmpty());
+ assertMockEndpointsSatisfied();
+ ap.stop();
+ }
+
+ @Test
+ public void testAggregationFailureDiscardedAfterPreCompletion() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a1");
+
+ AggregateProcessor ap = createProcessor();
+ ap.setDiscardOnAggregationFailure(true);
+ ap.start();
+
+ ap.process(createExchange("a1"));
+ Exchange bad = createExchange("START-fail");
+ ap.process(bad);
+
+ assertNull(bad.getException());
+ assertTrue(ap.getAggregationRepository().getKeys().isEmpty());
+ assertMockEndpointsSatisfied();
+ ap.stop();
+ }
+
+ private AggregateProcessor createProcessor() {
+ AsyncProcessor done = new
SendProcessor(context.getEndpoint("mock:result"));
+ // pre-completes the current group when a body starting with START
arrives
+ AggregationStrategy strategy = new AggregationStrategy() {
+ @Override
+ public boolean canPreComplete() {
+ return true;
+ }
+
+ @Override
+ public boolean preComplete(Exchange oldExchange, Exchange
newExchange) {
+ return oldExchange != null &&
newExchange.getIn().getBody(String.class).startsWith("START");
+ }
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange
newExchange) {
+ String body = newExchange.getIn().getBody(String.class);
+ if (body.endsWith("fail")) {
+ throw new IllegalArgumentException("Cannot aggregate " +
body);
+ }
+ if (oldExchange == null) {
+ return newExchange;
+ }
+
oldExchange.getIn().setBody(oldExchange.getIn().getBody(String.class) + "+" +
body);
+ return oldExchange;
+ }
+ };
+ return new AggregateProcessor(context, done, header("id"), strategy,
executorService, true);
+ }
+
+ private Exchange createExchange(String body) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody(body);
+ exchange.getIn().setHeader("id", 1);
+ return exchange;
+ }
+
+ private static void process(AggregateProcessor ap, Exchange exchange) {
+ try {
+ ap.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ fail("Timeout waiting for latch");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ fail("Interrupted");
+ }
+ }
+}
