shashank created CAMEL-26642:
--------------------------------
Summary: camel-core - Aggregate EIP: a recover pass that outlives
a route restart can send a recovered exchange twice
Key: CAMEL-26642
URL: https://issues.apache.org/jira/browse/CAMEL-26642
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
Assignee: shashank
Found while checking the processors that start their own background tasks in
doStart for work that goes on after a quick stop and start (same family as
CAMEL-26634, the batch Resequencer). Shown with deterministic tests (latches in
the repository's {{scan}} and {{recover}}), no sleeps.
{{AggregateProcessor.doStop}} shuts down the executor of the recover task
({{RecoverableAggregationRepository}} with {{useRecovery}}, such as the JDBC,
LevelDB or key-value repositories) but does not wait for a recover pass that is
running. A pass checks {{isRunAllowed()}} before each exchange, so a pass that
is in one long call during the stop (the {{scan}} of the repository, the
{{recover}} load of an exchange, a dead letter send) goes on after a quick
start, next to the recover task of the new start. Then:
* the two passes share the processor's
{{inProgressCompleteExchangesForRecoveryTask}} set and {{recoveryInProgress}}
flag, and each pass clears them when it ends;
* {{doStart}} creates a new aggregation lock, so the two passes no longer
exclude each other.
{noformat}
old pass: scan() ... (slow, many exchanges to recover)
admin: stopRoute; startRoute (doStop does not wait for the old
pass)
new pass: snapshot = {} (Z not in progress), scan() ...
old pass: scan -> {Z}; recover(Z) and send it; ends: recoveryInProgress =
false, snapshot cleared
new pass: scan -> {Z} (not confirmed yet); the snapshot is empty -> recover(Z)
and send it again
{noformat}
or, with the old pass in {{recover(Z)}} across the restart: the new pass (new
lock) recovers and sends Z, then the old pass sends the copy it loaded. Either
way Z is sent twice. On the second path the old pass also unlocks the lock of
the new start ({{finally lock.unlock()}} reads the field again), which throws
an {{IllegalMonitorStateException}}.
Without a restart the passes never overlap (fixed delay on one thread), so the
in-progress check of CAMEL-24943 holds.
Ways to get there: a stop + start or restart of the route (JMX, control bus,
CLI, route policies, the supervising route controller) while a recover pass
runs. A pass can take long when many exchanges wait to be recovered (one
{{recover}} read per exchange), which is when recovery matters.
Expected: a recover pass of an earlier start ends at the stop, and an exchange
is recovered and sent once.
Suggested fix: run one recover pass at a time
({{recoveryInProgress.compareAndSet}}: a pass skips while another runs), and
unlock the lock that the pass locked. The same code is on 4.18.x and 4.14.x.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
