[
https://issues.apache.org/jira/browse/CAMEL-26642?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-26642 started by shashank.
----------------------------------------
> 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
> Priority: Major
>
> 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)
