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 06ce5afd122a CAMEL-24946: camel-support -
KeyValueAggregationRepository should store the completed exchange for recovery
06ce5afd122a is described below
commit 06ce5afd122ad92ac94c28823a2dd8b79f78088e
Author: smjain <[email protected]>
AuthorDate: Wed Sep 23 18:30:53 2026 +0530
CAMEL-24946: camel-support - KeyValueAggregationRepository should store the
completed exchange for recovery
KeyValueAggregationRepository.remove deleted the entry of the correlation
key
and moved the deleted value to the recovery store, ignoring the exchange
passed
to remove. When a group is completed by an incoming exchange
(completionSize,
completionPredicate, or the strategy), AggregateProcessor aggregates that
exchange in memory and does not add the result to the repository before it
removes the group, so the stored value was the group without the last
exchange(s). If routing the aggregated exchange failed, the recover task
re-delivered the group without its last message, and without the
AGGREGATED_CORRELATION_KEY property.
Store the exchange passed to remove in the recovery store instead, like
JdbcAggregationRepository does. The recovery copy is still only written when
the key was actually removed.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../AggregateKeyValueRepositoryRecoverTest.java | 66 ++++++++++++++++++++++
.../support/KeyValueAggregationRepositoryTest.java | 17 ++++++
.../support/KeyValueAggregationRepository.java | 7 ++-
3 files changed, 88 insertions(+), 2 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateKeyValueRepositoryRecoverTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateKeyValueRepositoryRecoverTest.java
new file mode 100644
index 000000000000..ca92ec5b0897
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateKeyValueRepositoryRecoverTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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 org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.BodyInAggregatingStrategy;
+import org.apache.camel.support.KeyValueAggregationRepository;
+import org.junit.jupiter.api.Test;
+
+/**
+ * An aggregated exchange recovered from a {@link
KeyValueAggregationRepository} must be the exchange that was
+ * completed, including the exchange that completed the group.
+ */
+public class AggregateKeyValueRepositoryRecoverTest extends ContextTestSupport
{
+
+ @Test
+ public void testRecoverGroupCompletedBySize() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("A+B+C", "A+B+C");
+ mock.message(0).header(Exchange.REDELIVERED).isNull();
+ mock.message(1).header(Exchange.REDELIVERED).isEqualTo(true);
+ // the first attempt fails, so the aggregated exchange is recovered
+ mock.whenExchangeReceived(1, exchange -> {
+ throw new IllegalArgumentException("Forced");
+ });
+
+ template.sendBodyAndHeader("direct:start", "A", "id", 1);
+ template.sendBodyAndHeader("direct:start", "B", "id", 1);
+ template.sendBodyAndHeader("direct:start", "C", "id", 1);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ KeyValueAggregationRepository repository = new
KeyValueAggregationRepository();
+ repository.setRecoveryInterval(100);
+
+ from("direct:start")
+ .aggregate(header("id"), new
BodyInAggregatingStrategy()).aggregationRepository(repository)
+ .completionSize(3)
+ .to("mock:result");
+ }
+ };
+ }
+}
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 4debfc8d8ad7..d40b85f84bc7 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
@@ -141,6 +141,23 @@ class KeyValueAggregationRepositoryTest {
assertThat(aggregationRepository.recover(camelContext,
exchangeId)).isNull();
}
+ @Test
+ void testRemoveStoresGivenExchangeForRecovery() {
+ aggregationRepository.add(camelContext, "key1", createExchange("A"));
+
+ // the group is completed by an exchange that is aggregated but not
added to the repository
+ Exchange aggregated = aggregationRepository.get(camelContext, "key1");
+ aggregated.getIn().setBody("A+B");
+ aggregated.setProperty(Exchange.AGGREGATED_CORRELATION_KEY, "key1");
+ aggregationRepository.remove(camelContext, "key1", aggregated);
+
+ Exchange recovered = aggregationRepository.recover(camelContext,
aggregated.getExchangeId());
+ assertThat(recovered).isNotNull();
+
assertThat(recovered.getExchangeId()).isEqualTo(aggregated.getExchangeId());
+ assertThat(recovered.getIn().getBody(String.class)).isEqualTo("A+B");
+ assertThat(recovered.getProperty(Exchange.AGGREGATED_CORRELATION_KEY,
String.class)).isEqualTo("key1");
+ }
+
@Test
void testScanReturnsCompletedExchangeIds() {
Exchange ex1 = createExchange("One");
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 38a4aaf24301..ccf651a8c266 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
@@ -124,9 +124,12 @@ public class KeyValueAggregationRepository extends
ServiceSupport
LOG.trace("Removing an Exchange with ID {} for key {}",
exchange.getExchangeId(), key);
DefaultExchangeHolder holder = (DefaultExchangeHolder)
repository.delete(AGGREGATE_PREFIX + key);
if (useRecovery && holder != null) {
- // Store under the exchangeId for potential recovery
+ // Store the given exchange under the exchangeId for potential
recovery. The removed holder is not used as
+ // 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(),
holder);
+ repository.put(COMPLETED_PREFIX + exchange.getExchangeId(),
+ DefaultExchangeHolder.marshal(exchange, true,
allowSerializedHeaders));
}
}
