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 deadcabda307 CAMEL-25709: camel-core - Aggregate EIP: try the
completion timeout of a group again when the repository fails (#27989)
deadcabda307 is described below
commit deadcabda3076a77728b70d8bc1aa6ef9f0dcbad
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 14:42:59 2026 +0530
CAMEL-25709: camel-core - Aggregate EIP: try the completion timeout of a
group again when the repository fails (#27989)
The timeout map removes an expired entry before it calls the aggregator's
listener, and DefaultTimeoutMap.emitEvent ignores an exception of a listener
without logging it. When the aggregation repository failed in get or remove
while the timeout checker completed a group (only OptimisticLockingException
was handled), the group stayed in the repository and was never completed by
its timeout, unless another message for the key arrived or the route was
restarted, and nothing was logged.
AggregationTimeoutMap.onEviction now catches such an exception, logs a WARN
and puts the key back into the timeout map (putIfAbsent, so an entry of a
new
message for the key is kept) to be tried again after the completion timeout
checker interval. A group that was already removed is not found by the next
try, which then does nothing.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../processor/aggregate/AggregateProcessor.java | 49 ++++++
.../AggregateTimeoutRepositoryExceptionTest.java | 186 +++++++++++++++++++++
2 files changed, 235 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 93392d6c66a3..94228e2714b8 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
@@ -1556,12 +1556,35 @@ public class AggregateProcessor extends
BaseProcessorSupport
*/
private final class AggregationTimeoutMap extends
DefaultTimeoutMap<String, String> {
+ // the keys whose completion by timeout failed and is tried again:
only the first failure is logged at WARN, so
+ // that a failure that does not go away does not log a WARN at every
check
+ private final Set<String> retriedKeys = ConcurrentHashMap.newKeySet();
+
private AggregationTimeoutMap(ScheduledExecutorService executor, long
requestMapPollTimeMillis) {
// do NOT use locking on the timeout map as this aggregator has
its own shared lock we will use instead
super(executor, requestMapPollTimeMillis, optimisticLocking);
addListener(this::onEviction);
}
+ @Override
+ public String put(String key, String value, long timeoutMillis) {
+ // a new message or group for the key: a failure of its timeout is
a new one
+ forgetRetry(key);
+ return super.put(key, value, timeoutMillis);
+ }
+
+ @Override
+ public String remove(String key) {
+ forgetRetry(key);
+ return super.remove(key);
+ }
+
+ private void forgetRetry(String key) {
+ if (!retriedKeys.isEmpty()) {
+ retriedKeys.remove(key);
+ }
+ }
+
@Override
protected void purge() {
// wait for lock to be created
@@ -1580,6 +1603,32 @@ public class AggregateProcessor extends
BaseProcessorSupport
if (type != Listener.Type.Evict) {
return;
}
+ try {
+ doOnEviction(key, exchangeId);
+ forgetRetry(key);
+ } catch (Exception e) {
+ // the entry is no longer in the timeout map, and the timeout
map ignores errors of its listeners, so
+ // without a new entry a group the repository could not read
or remove would never time out (an entry
+ // that a new message for the key has put in the meantime is
kept)
+ long delay = getCompletionTimeoutCheckerInterval();
+ if (retriedKeys.add(key)) {
+ LOG.warn("Error completing the group with correlation key:
{} by its completion timeout,"
+ + " will try again every {} ms (further failures
for this key are logged at DEBUG)."
+ + " Caused by: {}",
+ key, delay, e.toString());
+ LOG.debug("Error completing the group with correlation
key: {} by its completion timeout", key, e);
+ } else {
+ LOG.debug("Error completing the group with correlation
key: {} by its completion timeout again,"
+ + " will try again in {} ms",
+ key, delay, e);
+ }
+ if (isRunAllowed()) {
+ putIfAbsent(key, exchangeId, delay);
+ }
+ }
+ }
+
+ private void doOnEviction(String key, String exchangeId) {
log.debug("Completion timeout triggered for correlation key: {}",
key);
// with optimistic locking the exchange id in the entry can belong
to a group that has been completed
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
new file mode 100644
index 000000000000..7bdc3ca2dfaf
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateTimeoutRepositoryExceptionTest.java
@@ -0,0 +1,186 @@
+/*
+ * 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.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.log.ConsumingAppender;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.aggregate.AggregateProcessor;
+import org.apache.camel.processor.aggregate.MemoryAggregationRepository;
+import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy;
+import org.apache.camel.spi.AggregationRepository;
+import org.apache.camel.support.service.ServiceSupport;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A group that is due for its completion timeout must still be completed when
the aggregation repository fails once (a
+ * database that is not reachable for a moment) while the timeout checker
completes it. The group stays in the
+ * repository, so it must be completed by a later check, and not be left in
the repository for ever.
+ */
+public class AggregateTimeoutRepositoryExceptionTest extends
ContextTestSupport {
+
+ private static final String LOGGER = AggregateProcessor.class.getName();
+
+ private final FailOnceRepository repository = new FailOnceRepository();
+
+ @AfterEach
+ public void removeAppender() {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ ctx.getConfiguration().removeLogger(LOGGER);
+ ctx.updateLoggers();
+ }
+
+ @Test
+ public void testRepeatedFailuresOfAGroupLogOneWarn() throws Exception {
+ // the tests in this module run in parallel and log through the same
logger: keep the lines of this key only
+ String key = "warn-once-" + context.getName();
+ Queue<String> warnings = new ConcurrentLinkedQueue<>();
+ ConsumingAppender.newAppender(LOGGER,
"AggregateTimeoutRepositoryExceptionTest", Level.WARN, event -> {
+ String message = event.getMessage().getFormattedMessage();
+ if (event.getLevel() == Level.WARN && message.contains(key)) {
+ warnings.add(message);
+ }
+ });
+
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("B");
+
+ template.sendBodyAndHeader("direct:start", "A", "id", key);
+ template.sendBodyAndHeader("direct:start", "B", "id", key);
+ // the next three reads of the group are the ones of the timeout
checker
+ repository.failingGets.set(3);
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ assertEquals(3, repository.failures.get(), "the repository should have
failed three times");
+ assertEquals(1, warnings.size(), "only the first failure is logged at
WARN: " + warnings);
+ String warning = List.copyOf(warnings).get(0);
+ assertTrue(warning.contains("java.lang.IllegalStateException:
Simulated repository failure in get"),
+ "the WARN names the type of the exception: " + warning);
+ }
+
+ @Test
+ public void testTimeoutCompletesGroupAfterFailedGet() throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("B");
+
+ template.sendBodyAndHeader("direct:start", "A", "id", "1");
+ template.sendBodyAndHeader("direct:start", "B", "id", "1");
+ // the next read of the group is the one of the timeout checker
+ repository.failNextGet.set(true);
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ assertEquals(1, repository.failures.get(), "the repository should have
failed once");
+ }
+
+ @Test
+ public void testTimeoutCompletesGroupAfterFailedRemove() throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("B");
+
+ template.sendBodyAndHeader("direct:start", "A", "id", "1");
+ template.sendBodyAndHeader("direct:start", "B", "id", "1");
+ // the next removal of the group is the one of the timeout checker
+ repository.failNextRemove.set(true);
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ assertEquals(1, repository.failures.get(), "the repository should have
failed once");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .aggregate(header("id"), new
UseLatestAggregationStrategy())
+ .aggregationRepository(repository)
+
.completionTimeout(200).completionTimeoutCheckerInterval(50)
+ .to("mock:result");
+ }
+ };
+ }
+
+ private static class FailOnceRepository extends ServiceSupport implements
AggregationRepository {
+
+ private final MemoryAggregationRepository delegate = new
MemoryAggregationRepository();
+ private final AtomicBoolean failNextGet = new AtomicBoolean();
+ private final AtomicBoolean failNextRemove = new AtomicBoolean();
+ private final AtomicInteger failingGets = new AtomicInteger();
+ private final AtomicInteger failures = new AtomicInteger();
+
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange
exchange) {
+ return delegate.add(camelContext, key, exchange);
+ }
+
+ @Override
+ public Exchange get(CamelContext camelContext, String key) {
+ if (failNextGet.compareAndSet(true, false) ||
failingGets.getAndUpdate(n -> Math.max(0, n - 1)) > 0) {
+ failures.incrementAndGet();
+ throw new IllegalStateException("Simulated repository failure
in get");
+ }
+ return delegate.get(camelContext, key);
+ }
+
+ @Override
+ public void remove(CamelContext camelContext, String key, Exchange
exchange) {
+ if (failNextRemove.compareAndSet(true, false)) {
+ failures.incrementAndGet();
+ throw new IllegalStateException("Simulated repository failure
in remove");
+ }
+ delegate.remove(camelContext, key, exchange);
+ }
+
+ @Override
+ public void confirm(CamelContext camelContext, String exchangeId) {
+ delegate.confirm(camelContext, exchangeId);
+ }
+
+ @Override
+ public Set<String> getKeys() {
+ return delegate.getKeys();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ delegate.start();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ delegate.stop();
+ }
+ }
+}
