Federico Mariani created CAMEL-24141:
----------------------------------------
Summary: camel-core - Aggregate EIP recovery and
completionInterval background tasks are permanently cancelled by the first
exception
Key: CAMEL-24141
URL: https://issues.apache.org/jira/browse/CAMEL-24141
Project: Camel
Issue Type: Bug
Components: camel-core, eip
Affects Versions: 4.21.0
Reporter: Federico Mariani
Two of the Aggregate EIP's periodic background tasks are scheduled directly on
a {{ScheduledExecutorService}} without any exception guard. Per
{{scheduleWithFixedDelay}}/{{scheduleAtFixedRate}} semantics, the first
exception that escapes {{run()}} *permanently cancels all subsequent
executions*:
# *RecoverTask* ({{AggregateProcessor.java:1449-1557}}, scheduled at :1635):
{{recoverable.scan(camelContext)}}, {{recover(...)}}, {{confirm(...)}} and
{{onSubmitCompletion(...)}} are all outside any try/catch (only the dead-letter
{{send}} is wrapped). A single transient failure — e.g. a momentary DB outage
while {{JdbcAggregationRepository.scan()}} runs — permanently disables recovery
until the route is restarted, with nothing but a one-off stack trace in the
logs. For persistent aggregation this means completed-but-unconfirmed exchanges
are never redelivered: effective data loss.
# *AggregationIntervalTask* ({{AggregateProcessor.java:1384-1437}}, scheduled
at :1666): {{aggregationRepository.getKeys()}} is called outside any try/catch,
and inside the loop only {{OptimisticLockingException}} is caught. Any other
repository/strategy exception kills the fixed-rate task forever — the
aggregator keeps accepting and aggregating messages but *never completes any
group by completionInterval*: unbounded memory growth with a memory repository,
indefinitely stuck groups with a persistent one.
The project already knows this failure mode: {{DefaultTimeoutMap.purgeTask()}}
({{DefaultTimeoutMap.java:163-176}}) catches all exceptions with the comment
_"must catch and log exception otherwise the executor will not schedule next
purgeTask"_ — which is why the completionTimeout checker survives repository
hiccups but recovery and interval completion do not.
Related robustness issue on the same path: {{aggregateCompletionCounter}}
({{AggregateProcessor.java:924-952}}) does {{switch (completedBy)}} which NPEs
when the {{CamelAggregatedCompletedBy}} property is absent (e.g. a recovered
exchange from a repository that did not persist that property) — on the
recovery path that NPE is itself a task-killer via the missing guard above. A
null-safe default branch would fix it.
*Reproducers (attached, both fail on main):*
* {{AggregateRecoverTaskDiesOnExceptionIssueTest}} — recoverable repository
whose {{scan()}} throws once; the pending unconfirmed exchange is never
recovered afterwards.
* {{AggregateIntervalTaskDiesOnExceptionIssueTest}} — repository whose
{{getKeys()}} throws once; a group aggregated afterwards is never completed by
interval.
Suggested fix: wrap the body of both tasks in try/catch that routes to the
aggregator's {{ExceptionHandler}} (same pattern as
{{DefaultTimeoutMap.purgeTask}}), and make {{aggregateCompletionCounter}}
null-safe.
_Filed by Claude Code on behalf of Federico Mariani (fmariani)_
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
