You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ap...@apache.org on 2026/09/24 08:16:39 UTC
(camel) branch main updated (b7b1a5bc173c -> 9d8866747c35)
This is an automated email from the ASF dual-hosted git repository.
apupier pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
from b7b1a5bc173c Bump org.hibernate.validator:hibernate-validator
new 8e781863ab13 CAMEL-24941: camel-core - Aggregate EIP: check the closed correlation key again under the lock
new 4cde7ce99276 CAMEL-24941: camel-core - Make the new test class package-private
new 9d8866747c35 CAMEL-24941: camel-core - Assert that the producer threads of the new test completed
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../processor/aggregate/AggregateProcessor.java | 5 +
.../AggregateClosedCorrelationKeyRaceTest.java | 222 +++++++++++++++++++++
2 files changed, 227 insertions(+)
create mode 100644 core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
(camel) 01/03: CAMEL-24941: camel-core - Aggregate EIP: check the closed correlation key again under the lock
Posted by ap...@apache.org.
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
commit 8e781863ab13d1995916932ffcc7a146282c89c4
Author: smjain <49...@users.noreply.github.com>
AuthorDate: Wed Sep 23 18:01:37 2026 +0530
CAMEL-24941: camel-core - Aggregate EIP: check the closed correlation key again under the lock
With closeCorrelationKeyOnCompletion the closed-key check in
AggregateProcessor.doProcess ran before the aggregation lock was taken, and
with optimistic locking only once before the retry loop. An exchange that
passed the check while another thread completed the group of the same key
(by size, predicate, timeout or force completion) waited for the lock and was
then aggregated into a new group for the already closed key. That group was
completed again later, so the key produced two aggregated exchanges instead of
one, and the late exchange was not rejected with ClosedCorrelationKeyException.
Check the closed correlation keys again after the lock has been taken, which
also runs on every optimistic locking retry. The first check is kept as a fast
path.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
.../processor/aggregate/AggregateProcessor.java | 5 +
.../AggregateClosedCorrelationKeyRaceTest.java | 218 +++++++++++++++++++++
2 files changed, 223 insertions(+)
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 1f5d8b856f41..289360076d45 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
@@ -451,6 +451,11 @@ public class AggregateProcessor extends BaseProcessorSupport
List<Exchange> aggregated = null;
lock.lock();
try {
+ // check again under the lock (and on every optimistic locking retry), as the key may have been closed
+ // by a completion that happened after the check in doProcess(Exchange, AsyncCallback)
+ if (closedCorrelationKeys != null && closedCorrelationKeys.containsKey(key)) {
+ throw new ClosedCorrelationKeyException(key, exchange);
+ }
aggregated = doAggregation(key, copy);
} catch (CamelExchangeException e) {
exchange.setException(e);
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
new file mode 100644
index 000000000000..61db6a5e0fad
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.AsyncCallback;
+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.ClosedCorrelationKeyException;
+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.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * An exchange that passed the closed correlation key check before the group of its key was completed (and the key
+ * closed) must not start a new group for the closed key.
+ */
+public class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
+
+ private final CountDownLatch inAggregate = new CountDownLatch(1);
+ private final CountDownLatch releaseAggregate = new CountDownLatch(1);
+ private final CountDownLatch passedClosedKeyCheck = 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 {
+ releaseAggregate.countDown();
+ executorService.shutdownNow();
+ super.tearDown();
+ }
+
+ @Test
+ public void testExchangeWaitingForLockWhileKeyIsClosed() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("A+B");
+
+ AggregateProcessor ap = createProcessor(false);
+ ap.setAggregationRepository(new MemoryAggregationRepository());
+ ap.start();
+
+ ap.process(createExchange("A"));
+
+ // B completes the group, and is paused in the aggregation strategy while it holds the aggregation lock
+ Exchange b = createExchange("B");
+ Thread producerB = new Thread(() -> process(ap, b), "producer-B");
+ producerB.start();
+ await(inAggregate);
+
+ // C passes the closed correlation key check (the key is not closed yet) and waits for the lock
+ Exchange c = createExchange("C");
+ Thread producerC = new Thread(() -> process(ap, c), "producer-C");
+ producerC.start();
+ await(passedClosedKeyCheck);
+
+ // B completes the group and closes the key, then C gets the lock
+ releaseAggregate.countDown();
+ producerB.join(10000);
+ producerC.join(10000);
+
+ assertTrue(c.getException() instanceof ClosedCorrelationKeyException,
+ "Expected ClosedCorrelationKeyException but was: " + c.getException());
+ assertTrue(ap.getAggregationRepository().getKeys().isEmpty(),
+ "No new group should be started for the closed key, but was: " + ap.getAggregationRepository().getKeys());
+ assertMockEndpointsSatisfied();
+
+ ap.stop();
+ }
+
+ @Test
+ public void testOptimisticLockingRetryAfterKeyIsClosed() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("A+B");
+
+ // pauses the first get of C, so B can complete the group and close the key before C continues
+ AtomicBoolean pauseGet = new AtomicBoolean(true);
+ MemoryAggregationRepository repository = new MemoryAggregationRepository(true) {
+ @Override
+ public Exchange get(CamelContext camelContext, String key) {
+ Exchange answer = super.get(camelContext, key);
+ if (Thread.currentThread().getName().equals("producer-C") && pauseGet.getAndSet(false)) {
+ passedClosedKeyCheck.countDown();
+ AggregateClosedCorrelationKeyRaceTest.await(releaseAggregate);
+ }
+ return answer;
+ }
+ };
+
+ AggregateProcessor ap = createProcessor(true);
+ 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("A"));
+
+ // C reads the group [A] and is paused
+ Exchange c = createExchange("C");
+ Thread producerC = new Thread(() -> process(ap, c), "producer-C");
+ producerC.start();
+ await(passedClosedKeyCheck);
+
+ // B completes the group [A, B] and closes the key
+ ap.process(createExchange("B"));
+
+ // C fails to remove the group it read (optimistic locking) and is retried
+ releaseAggregate.countDown();
+ producerC.join(10000);
+
+ assertTrue(c.getException() instanceof ClosedCorrelationKeyException,
+ "Expected ClosedCorrelationKeyException but was: " + c.getException());
+ assertTrue(ap.getAggregationRepository().getKeys().isEmpty(),
+ "No new group should be started for the closed key, but was: " + ap.getAggregationRepository().getKeys());
+ assertMockEndpointsSatisfied();
+
+ ap.stop();
+ }
+
+ private AggregateProcessor createProcessor(boolean optimistic) {
+ AsyncProcessor done = new SendProcessor(context.getEndpoint("mock:result"));
+ AggregationStrategy strategy = (oldExchange, newExchange) -> {
+ String body = newExchange.getIn().getBody(String.class);
+ if (!optimistic && "B".equals(body)) {
+ inAggregate.countDown();
+ await(releaseAggregate);
+ }
+ if (oldExchange == null) {
+ return newExchange;
+ }
+ oldExchange.getIn().setBody(oldExchange.getIn().getBody(String.class) + "+" + body);
+ return oldExchange;
+ };
+
+ AggregateProcessor ap = new AggregateProcessor(context, done, header("id"), strategy, executorService, true) {
+ @Override
+ protected boolean doProcess(Exchange exchange, String key, AsyncCallback callback, boolean sync) {
+ // called after the closed correlation key check in process(Exchange, AsyncCallback)
+ if (!optimistic && "C".equals(exchange.getIn().getBody(String.class))) {
+ passedClosedKeyCheck.countDown();
+ }
+ return super.doProcess(exchange, key, callback, sync);
+ }
+ };
+ ap.setCompletionSize(2);
+ ap.setCloseCorrelationKeyOnCompletion(100);
+ return ap;
+ }
+
+ 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");
+ }
+ }
+}
(camel) 03/03: CAMEL-24941: camel-core - Assert that the producer threads of the new test completed
Posted by ap...@apache.org.
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
commit 9d8866747c352eba0fccc6f828821154310ee218
Author: smjain <49...@users.noreply.github.com>
AuthorDate: Thu Sep 24 11:45:05 2026 +0530
CAMEL-24941: camel-core - Assert that the producer threads of the new test completed
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
.../processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
index 35b28f65fb59..61e344a1efbb 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
@@ -39,6 +39,7 @@ 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.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -100,6 +101,8 @@ class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
releaseAggregate.countDown();
producerB.join(10000);
producerC.join(10000);
+ assertFalse(producerB.isAlive(), "producer-B did not complete within 10 s");
+ assertFalse(producerC.isAlive(), "producer-C did not complete within 10 s");
assertTrue(c.getException() instanceof ClosedCorrelationKeyException,
"Expected ClosedCorrelationKeyException but was: " + c.getException());
@@ -150,6 +153,7 @@ class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
// C fails to remove the group it read (optimistic locking) and is retried
releaseAggregate.countDown();
producerC.join(10000);
+ assertFalse(producerC.isAlive(), "producer-C did not complete within 10 s");
assertTrue(c.getException() instanceof ClosedCorrelationKeyException,
"Expected ClosedCorrelationKeyException but was: " + c.getException());
(camel) 02/03: CAMEL-24941: camel-core - Make the new test class package-private
Posted by ap...@apache.org.
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
commit 4cde7ce99276bb9b906adff6e0d0327fe826dc47
Author: smjain <49...@users.noreply.github.com>
AuthorDate: Thu Sep 24 07:17:14 2026 +0530
CAMEL-24941: camel-core - Make the new test class package-private
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
.../processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
index 61db6a5e0fad..35b28f65fb59 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateClosedCorrelationKeyRaceTest.java
@@ -46,7 +46,7 @@ import static org.junit.jupiter.api.Assertions.fail;
* An exchange that passed the closed correlation key check before the group of its key was completed (and the key
* closed) must not start a new group for the closed key.
*/
-public class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
+class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
private final CountDownLatch inAggregate = new CountDownLatch(1);
private final CountDownLatch releaseAggregate = new CountDownLatch(1);
@@ -74,7 +74,7 @@ public class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
}
@Test
- public void testExchangeWaitingForLockWhileKeyIsClosed() throws Exception {
+ void testExchangeWaitingForLockWhileKeyIsClosed() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedBodiesReceived("A+B");
@@ -111,7 +111,7 @@ public class AggregateClosedCorrelationKeyRaceTest extends ContextTestSupport {
}
@Test
- public void testOptimisticLockingRetryAfterKeyIsClosed() throws Exception {
+ void testOptimisticLockingRetryAfterKeyIsClosed() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedBodiesReceived("A+B");