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 93950e50adf0 CAMEL-24141: Aggregate EIP - guard background tasks
against exceptions
93950e50adf0 is described below
commit 93950e50adf0c62b7d24f012bac70e2f0a9f39dc
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Jul 17 16:00:59 2026 +0200
CAMEL-24141: Aggregate EIP - guard background tasks against exceptions
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../processor/aggregate/AggregateProcessor.java | 67 ++++++------
.../AggregateIntervalTaskExceptionTest.java | 112 +++++++++++++++++++++
2 files changed, 151 insertions(+), 28 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 96cadd4ba3b5..a9c8debbb09c 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
@@ -923,6 +923,9 @@ public class AggregateProcessor extends BaseProcessorSupport
private void aggregateCompletionCounter(Exchange exchange) {
String completedBy =
exchange.getProperty(ExchangePropertyKey.AGGREGATED_COMPLETED_BY, String.class);
+ if (completedBy == null) {
+ return;
+ }
switch (completedBy) {
case COMPLETED_BY_INTERVAL:
completedByInterval.incrementAndGet();
@@ -1396,40 +1399,45 @@ public class AggregateProcessor extends
BaseProcessorSupport
LOG.trace("Starting completion interval task");
- // trigger completion for all in the repository
- Set<String> keys = aggregationRepository.getKeys();
+ // must catch and log exception otherwise the executor will not
schedule next interval task
+ try {
+ // trigger completion for all in the repository
+ Set<String> keys = aggregationRepository.getKeys();
- if (keys != null && !keys.isEmpty()) {
- // must acquire the shared aggregation lock to be able to
trigger interval completion
- lock.lock();
- try {
- for (String key : keys) {
- boolean stolenInterval = false;
- Exchange exchange =
aggregationRepository.get(camelContext, key);
- if (exchange == null) {
- stolenInterval = true;
- } else {
- LOG.trace("Completion interval triggered for
correlation key: {}", key);
- // indicate it was completed by interval
-
exchange.setProperty(ExchangePropertyKey.AGGREGATED_COMPLETED_BY,
COMPLETED_BY_INTERVAL);
- try {
- Exchange answer = onCompletion(key, exchange,
exchange, false, false);
- if (answer != null) {
- onSubmitCompletion(key, answer);
- }
- } catch
(OptimisticLockingAggregationRepository.OptimisticLockingException e) {
+ if (keys != null && !keys.isEmpty()) {
+ // must acquire the shared aggregation lock to be able to
trigger interval completion
+ lock.lock();
+ try {
+ for (String key : keys) {
+ boolean stolenInterval = false;
+ Exchange exchange =
aggregationRepository.get(camelContext, key);
+ if (exchange == null) {
stolenInterval = true;
+ } else {
+ LOG.trace("Completion interval triggered for
correlation key: {}", key);
+ // indicate it was completed by interval
+
exchange.setProperty(ExchangePropertyKey.AGGREGATED_COMPLETED_BY,
COMPLETED_BY_INTERVAL);
+ try {
+ Exchange answer = onCompletion(key,
exchange, exchange, false, false);
+ if (answer != null) {
+ onSubmitCompletion(key, answer);
+ }
+ } catch
(OptimisticLockingAggregationRepository.OptimisticLockingException e) {
+ stolenInterval = true;
+ }
+ }
+ if (optimisticLocking && stolenInterval) {
+ LOG.debug(
+ "Another Camel instance has already
processed this interval aggregation for exchange with correlation id: {}",
+ key);
}
}
- if (optimisticLocking && stolenInterval) {
- LOG.debug(
- "Another Camel instance has already
processed this interval aggregation for exchange with correlation id: {}",
- key);
- }
+ } finally {
+ lock.unlock();
}
- } finally {
- lock.unlock();
}
+ } catch (Exception e) {
+ LOG.warn("Error during completion interval task. This
exception is ignored.", e);
}
LOG.trace("Completion interval task complete");
@@ -1549,6 +1557,9 @@ public class AggregateProcessor extends
BaseProcessorSupport
lock.unlock();
}
}
+ } catch (Exception e) {
+ // must catch and log exception otherwise the executor will
not schedule next recover task
+ LOG.warn("Error during recover task. This exception is
ignored.", e);
} finally {
recoveryInProgress.set(false);
inProgressCompleteExchangesForRecoveryTask.clear();
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
new file mode 100644
index 000000000000..67d3abac8b43
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.Set;
+import java.util.concurrent.TimeUnit;
+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.mock.MockEndpoint;
+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.junit.jupiter.api.Test;
+
+/**
+ * Test that the completion interval task survives an exception thrown by the
aggregation repository. Without the
+ * try/catch guard in AggregationIntervalTask.run(), a single RuntimeException
from getKeys() would permanently cancel
+ * the scheduled task.
+ */
+public class AggregateIntervalTaskExceptionTest extends ContextTestSupport {
+
+ @Test
+ public void testIntervalTaskSurvivesException() throws Exception {
+ getMockEndpoint("mock:result").expectedMinimumMessageCount(1);
+
+ // send messages that will be aggregated
+ template.sendBodyAndHeader("direct:start", "A", "id", "1");
+ template.sendBodyAndHeader("direct:start", "B", "id", "1");
+
+ // the first interval tick with data will hit the exception from
getKeys(),
+ // but the scheduler should survive and complete on a subsequent tick
+ MockEndpoint.assertIsSatisfied(context, 15, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .aggregate(header("id"), new
UseLatestAggregationStrategy())
+ .aggregationRepository(new FailOnceRepository())
+ .completionInterval(500)
+ .to("mock:result");
+ }
+ };
+ }
+
+ private static class FailOnceRepository extends ServiceSupport implements
AggregationRepository {
+
+ private final MemoryAggregationRepository delegate = new
MemoryAggregationRepository();
+ private final AtomicInteger getKeysCount = 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) {
+ return delegate.get(camelContext, key);
+ }
+
+ @Override
+ public void remove(CamelContext camelContext, String key, Exchange
exchange) {
+ delegate.remove(camelContext, key, exchange);
+ }
+
+ @Override
+ public void confirm(CamelContext camelContext, String exchangeId) {
+ delegate.confirm(camelContext, exchangeId);
+ }
+
+ @Override
+ public Set<String> getKeys() {
+ Set<String> keys = delegate.getKeys();
+ if (keys != null && !keys.isEmpty() &&
getKeysCount.incrementAndGet() == 1) {
+ throw new RuntimeException("Simulated repository failure");
+ }
+ return keys;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ delegate.start();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ delegate.stop();
+ }
+ }
+}
