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 61952f53597b CAMEL-25864: camel-core - an aggregator recovers only the
exchanges it completed in an auto-discovered KeyValueRepository (#28383)
61952f53597b is described below
commit 61952f53597b0c9dc61aae6118f2e56fe3558c2a
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 25 13:00:25 2026 +0530
CAMEL-25864: camel-core - an aggregator recovers only the exchanges it
completed in an auto-discovered KeyValueRepository (#28383)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/docs/keyValueRepository.adoc | 10 ++
.../modules/eips/pages/keyValueRepository.adoc | 10 ++
.../processor/aggregate/AggregateProcessor.java | 5 +-
...AutoDiscoveredKeyValueRepositorySharedTest.java | 101 +++++++++++++++++++++
.../support/KeyValueAggregationRepositoryTest.java | 28 ++++++
.../support/KeyValueAggregationRepository.java | 35 ++++++-
.../modules/ROOT/pages/key-value-repository.adoc | 10 ++
7 files changed, 193 insertions(+), 6 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
index c8398fe57026..119785e05086 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
@@ -1154,6 +1154,16 @@ store. For backends without native TTL (JDBC, JPA,
Kafka), aggregation-in-progre
persist until completion. Use a backend with durable storage (JDBC, JPA,
Infinispan) when
crash recovery of in-flight aggregations is required.
+NOTE: Every aggregator without an `aggregationRepository` wraps the same
auto-discovered
+`KeyValueRepository`, and keeps its completed exchanges for recovery under its
route id and
+aggregator id (`<routeId>/<id>`), so it recovers only its own. With a
persistent store, set explicit
+route and aggregator ids: Camel generates them (`route1`, `aggregate1`) when
they are not set, and
+they can change when routes are added or reordered between restarts. When
several aggregators use one
+`KeyValueRepository` through explicit `KeyValueAggregationRepository` beans,
create the beans with the
+`KeyValueAggregationRepository(repository, recoveryName)` constructor and a
different name for each
+aggregator. Without a name, the completed exchanges of all the aggregators are
kept under the same
+prefix, and each aggregator recovers, and sends again on its own route, the
exchanges of the others.
+
== See Also
* xref:cache-eip.adoc[Cache EIP]
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
index c8398fe57026..119785e05086 100644
---
a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
+++
b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
@@ -1154,6 +1154,16 @@ store. For backends without native TTL (JDBC, JPA,
Kafka), aggregation-in-progre
persist until completion. Use a backend with durable storage (JDBC, JPA,
Infinispan) when
crash recovery of in-flight aggregations is required.
+NOTE: Every aggregator without an `aggregationRepository` wraps the same
auto-discovered
+`KeyValueRepository`, and keeps its completed exchanges for recovery under its
route id and
+aggregator id (`<routeId>/<id>`), so it recovers only its own. With a
persistent store, set explicit
+route and aggregator ids: Camel generates them (`route1`, `aggregate1`) when
they are not set, and
+they can change when routes are added or reordered between restarts. When
several aggregators use one
+`KeyValueRepository` through explicit `KeyValueAggregationRepository` beans,
create the beans with the
+`KeyValueAggregationRepository(repository, recoveryName)` constructor and a
different name for each
+aggregator. Without a name, the completed exchanges of all the aggregators are
kept under the same
+prefix, and each aggregator recovers, and sends again on its own route, the
exchanges of the others.
+
== See Also
* xref:cache-eip.adoc[Cache EIP]
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 cb007e4f7ee1..f7cf84d3fcf9 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
@@ -1905,7 +1905,10 @@ public class AggregateProcessor extends
BaseProcessorSupport
// Auto-discover a KeyValueRepository from the registry and wrap it
KeyValueRepository kvRepo =
camelContext.getRegistry().findSingleByType(KeyValueRepository.class);
if (kvRepo != null) {
- setAggregationRepository(new
KeyValueAggregationRepository(kvRepo));
+ // every aggregator without a repository discovers the same
KeyValueRepository, so each one keeps its
+ // completed exchanges under its own name and recovers only
those
+ String recoveryName = routeId != null && id != null ? routeId
+ "/" + id : null;
+ setAggregationRepository(new
KeyValueAggregationRepository(kvRepo, recoveryName));
LOG.info("Auto-discovered KeyValueRepository from registry,
wrapping as AggregationRepository");
} else {
aggregationRepository = new
MemoryAggregationRepository(optimisticLocking);
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateAutoDiscoveredKeyValueRepositorySharedTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateAutoDiscoveredKeyValueRepositorySharedTest.java
new file mode 100644
index 000000000000..8b73d9730fb0
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateAutoDiscoveredKeyValueRepositorySharedTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.BodyInAggregatingStrategy;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.support.MemoryKeyValueRepository;
+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.assertTrue;
+
+/**
+ * Every aggregator without a repository auto-discovers the single {@code
KeyValueRepository} of the registry. An
+ * aggregator must only recover the exchanges it completed itself, not an
exchange that another aggregator completed and
+ * is still processing.
+ */
+class AggregateAutoDiscoveredKeyValueRepositorySharedTest extends
ContextTestSupport {
+
+ private final MemoryKeyValueRepository kvRepo = new
MemoryKeyValueRepository();
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ protected Registry createCamelRegistry() throws Exception {
+ Registry registry = super.createCamelRegistry();
+ registry.bind("kvRepo", kvRepo);
+ return registry;
+ }
+
+ @AfterEach
+ void releaseHeldExchange() {
+ // also when the test fails before, so that the held exchange does not
keep the shutdown waiting
+ release.countDown();
+ }
+
+ @Test
+ void testOtherAggregatorDoesNotRecoverExchangeInProgress() throws
Exception {
+ MockEndpoint orders = getMockEndpoint("mock:orders");
+ orders.expectedBodiesReceived("A+B");
+ MockEndpoint invoices = getMockEndpoint("mock:invoices");
+ invoices.expectedMessageCount(0);
+ // the recover checkers run 1 second after start and then every 5
seconds
+ invoices.setAssertPeriod(7000);
+
+ template.sendBodyAndHeader("direct:orders", "A", "id", 1);
+ // the group completes, and the aggregated exchange waits in the
orders route
+ Future<Object> completing =
template.asyncRequestBodyAndHeader("direct:orders", "B", "id", 1);
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertTrue(kvRepo.keys().stream().anyMatch(k -> k.startsWith("completed"))));
+
+ // the aggregator of the invoices route must not send the exchange of
the orders route
+ invoices.assertIsSatisfied();
+
+ release.countDown();
+ completing.get(5, TimeUnit.SECONDS);
+ orders.assertIsSatisfied();
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertTrue(kvRepo.keys().isEmpty()));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // no aggregationRepository: both aggregators auto-discover
the KeyValueRepository of the registry
+ from("direct:orders").routeId("orders")
+ .aggregate(header("id"), new
BodyInAggregatingStrategy())
+ .completionSize(2)
+ .process(e -> release.await(20, TimeUnit.SECONDS))
+ .to("mock:orders");
+
+ from("direct:invoices").routeId("invoices")
+ .aggregate(header("id"), new
BodyInAggregatingStrategy())
+ .completionSize(2)
+ .to("mock:invoices");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/support/KeyValueAggregationRepositoryTest.java
b/core/camel-core/src/test/java/org/apache/camel/support/KeyValueAggregationRepositoryTest.java
index d40b85f84bc7..63157cfb31f3 100644
---
a/core/camel-core/src/test/java/org/apache/camel/support/KeyValueAggregationRepositoryTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/support/KeyValueAggregationRepositoryTest.java
@@ -223,6 +223,34 @@ class KeyValueAggregationRepositoryTest {
assertThat(created.getRepository()).isSameAs(kvRepository);
}
+ @Test
+ void testRecoveryNameKeepsCompletedExchangesApart() throws Exception {
+ KeyValueAggregationRepository orders = new
KeyValueAggregationRepository(kvRepository, "orders/aggregate1");
+ KeyValueAggregationRepository invoices = new
KeyValueAggregationRepository(kvRepository, "invoices/aggregate2");
+ orders.setCamelContext(camelContext);
+ invoices.setCamelContext(camelContext);
+ orders.start();
+ invoices.start();
+
+ Exchange order = createExchange("order");
+ orders.add(camelContext, "1", order);
+ orders.remove(camelContext, "1", order);
+ Exchange invoice = createExchange("invoice");
+ invoices.add(camelContext, "2", invoice);
+ invoices.remove(camelContext, "2", invoice);
+
+ // each adapter only scans and recovers the exchanges it completed,
and so does an adapter without a name
+
assertThat(orders.scan(camelContext)).containsExactly(order.getExchangeId());
+
assertThat(invoices.scan(camelContext)).containsExactly(invoice.getExchangeId());
+ assertThat(aggregationRepository.scan(camelContext)).isEmpty();
+ assertThat(orders.recover(camelContext,
order.getExchangeId()).getIn().getBody(String.class)).isEqualTo("order");
+ assertThat(orders.recover(camelContext,
invoice.getExchangeId())).isNull();
+
+ orders.confirm(camelContext, order.getExchangeId());
+ assertThat(orders.scan(camelContext)).isEmpty();
+
assertThat(invoices.scan(camelContext)).containsExactly(invoice.getExchangeId());
+ }
+
private Exchange createExchange(String body) {
Exchange exchange = new DefaultExchange(camelContext);
exchange.getIn().setBody(body);
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
index ccf651a8c266..393492249da2 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/KeyValueAggregationRepository.java
@@ -69,6 +69,8 @@ public class KeyValueAggregationRepository extends
ServiceSupport
private static final String COMPLETED_PREFIX = "completed:";
private KeyValueRepository repository;
+ // the completed exchanges kept for recovery are stored under this prefix
+ private String completedPrefix = COMPLETED_PREFIX;
private CamelContext camelContext;
private boolean useRecovery = true;
private String deadLetterUri;
@@ -96,6 +98,29 @@ public class KeyValueAggregationRepository extends
ServiceSupport
this.repository = repository;
}
+ /**
+ * Creates an aggregation repository adapter backed by the given key-value
repository, which keeps the completed
+ * exchanges for recovery under the given name.
+ * <p/>
+ * Use this when several aggregators share the same key-value repository:
the recover task of an aggregator scans
+ * all the completed exchanges of its repository and sends them again, so
an aggregator would otherwise recover, and
+ * send a second time on its own route, an exchange that another
aggregator completed and is still processing.
+ *
+ * @param repository the underlying key-value repository; must not be
{@code null}
+ * @param recoveryName the name under which the completed exchanges are
kept, such as the route id and the id of the
+ * aggregator; {@code null} keeps them under the
default prefix shared by all the adapters
+ * without a name. The name should not contain {@code
]:}, as the keys of such a name can start
+ * with the prefix of another name (the keys of {@code
a]:b} start with the prefix of {@code a})
+ * @since 4.24
+ */
+ public KeyValueAggregationRepository(KeyValueRepository repository, String
recoveryName) {
+ this(repository);
+ if (recoveryName != null) {
+ // not under the default prefix, so that an adapter without a name
does not scan them
+ this.completedPrefix = "completed[" + recoveryName + "]:";
+ }
+ }
+
/**
* Creates a new {@link KeyValueAggregationRepository} wrapping the given
{@link KeyValueRepository}.
*
@@ -128,7 +153,7 @@ public class KeyValueAggregationRepository extends
ServiceSupport
// it can be older than the exchange: when a group is completed by
an incoming exchange, the last
// aggregated exchange is not added to the repository before it is
removed.
LOG.trace("Moving Exchange with ID {} to completed (pending
confirmation)", exchange.getExchangeId());
- repository.put(COMPLETED_PREFIX + exchange.getExchangeId(),
+ repository.put(completedPrefix + exchange.getExchangeId(),
DefaultExchangeHolder.marshal(exchange, true,
allowSerializedHeaders));
}
}
@@ -136,7 +161,7 @@ public class KeyValueAggregationRepository extends
ServiceSupport
@Override
public void confirm(CamelContext camelContext, String exchangeId) {
LOG.trace("Confirming an Exchange with ID {}", exchangeId);
- repository.delete(COMPLETED_PREFIX + exchangeId);
+ repository.delete(completedPrefix + exchangeId);
}
@Override
@@ -152,8 +177,8 @@ public class KeyValueAggregationRepository extends
ServiceSupport
public Set<String> scan(CamelContext camelContext) {
LOG.trace("Scanning for exchanges to recover in {} context",
camelContext.getName());
Set<String> scanned = repository.keys().stream()
- .filter(k -> k.startsWith(COMPLETED_PREFIX))
- .map(k -> k.substring(COMPLETED_PREFIX.length()))
+ .filter(k -> k.startsWith(completedPrefix))
+ .map(k -> k.substring(completedPrefix.length()))
.collect(Collectors.toUnmodifiableSet());
LOG.trace("Found {} keys for exchanges to recover in {} context",
scanned.size(), camelContext.getName());
return scanned;
@@ -165,7 +190,7 @@ public class KeyValueAggregationRepository extends
ServiceSupport
if (!useRecovery) {
return null;
}
- return unmarshallExchange(camelContext, (DefaultExchangeHolder)
repository.get(COMPLETED_PREFIX + exchangeId));
+ return unmarshallExchange(camelContext, (DefaultExchangeHolder)
repository.get(completedPrefix + exchangeId));
}
@Override
diff --git a/docs/user-manual/modules/ROOT/pages/key-value-repository.adoc
b/docs/user-manual/modules/ROOT/pages/key-value-repository.adoc
index 1b7e44909e35..1d5040b5765b 100644
--- a/docs/user-manual/modules/ROOT/pages/key-value-repository.adoc
+++ b/docs/user-manual/modules/ROOT/pages/key-value-repository.adoc
@@ -318,6 +318,16 @@ With this single bean in the registry:
No explicit wiring is needed.
+NOTE: Every aggregator without an `aggregationRepository` wraps the same
auto-discovered
+`KeyValueRepository`, and keeps its completed exchanges for recovery under its
route id and
+aggregator id (`<routeId>/<id>`), so it recovers only its own. With a
persistent store, set explicit
+route and aggregator ids: Camel generates them (`route1`, `aggregate1`) when
they are not set, and
+they can change when routes are added or reordered between restarts. When
several aggregators use one
+`KeyValueRepository` through explicit `KeyValueAggregationRepository` beans,
create the beans with the
+`KeyValueAggregationRepository(repository, recoveryName)` constructor and a
different name for each
+aggregator. Without a name, the completed exchanges of all the aggregators are
kept under the same
+prefix, and each aggregator recovers, and sends again on its own route, the
exchanges of the others.
+
=== Explicit wiring
You can also wire the backend explicitly when you need more control:
