[
https://issues.apache.org/jira/browse/CAMEL-24943?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-24943.
---------------------------------
Resolution: Fixed
Merged to main for 4.23.0 via https://github.com/apache/camel/pull/26788
(commit fe67aa8989d8196c7eb944d9cb61ba07a84fe43c).
_Claude Code on behalf of davsclaus_
> Aggregate EIP: the recover task can re-deliver an aggregated exchange that is
> being completed normally, which sends it twice
> ----------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24943
> URL: https://issues.apache.org/jira/browse/CAMEL-24943
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Priority: Minor
> Fix For: 4.23.0
>
>
> With a {{RecoverableAggregationRepository}} (JDBC, Redis, Hazelcast,
> Infinispan, {{KeyValueAggregationRepository}}, ...) an aggregated exchange
> can be sent twice even though nothing failed: once by the thread that
> completed the group and once by the recover task.
> When an incoming exchange completes a group, {{AggregateProcessor}} does the
> following:
> # Under the aggregation lock, {{onCompletion}} calls
> {{aggregationRepository.remove(...)}}. A recoverable repository moves the
> exchange to its recovery (completed) store there.
> # The lock is released.
> # {{onSubmitCompletion}} adds the exchange id to
> {{inProgressCompleteExchanges}} and sends the exchange.
> If the recover task scans the repository and checks the id between steps 2
> and 3, the id is not "in progress", so the task recovers the exchange and
> submits it. Then the completing thread submits it as well.
> Observed with {{completionSize(2)}}, recovery interval 100 ms, and the
> completing thread paused right after it releases the aggregation lock:
> {noformat}
> producer paused after unlock; completed store =
> [2A3CFA493BBF252-0000000000000001], in progress = 0
> recover task re-delivered the exchange while producer paused: true
> A+B id=2A3CFA493BBF252-0000000000000001 redelivered=true
> A+B id=2A3CFA493BBF252-0000000000000001 redelivered=null
> aggregated exchange delivered 2 times (expected 1)
> {noformat}
> Without the pause the exchange is delivered once. The same happens with
> {{KeyValueAggregationRepository}} with its default settings.
> With {{optimisticLocking}} the aggregation lock is a {{NoLock}}, so the
> recover task is not serialized with any completion path. A group completed by
> {{completionTimeout}} is also delivered twice if the recover task runs
> between the timeout checker's {{remove}} and its {{onSubmitCompletion}}:
> {noformat}
> A+B id=CCDAE062CAC9981-0000000000000001 redelivered=true completedBy=timeout
> A+B id=CCDAE062CAC9981-0000000000000001 redelivered=null completedBy=timeout
> {noformat}
> This is a remaining case of the race addressed by CAMEL-6097 and CAMEL-8010.
> Those fixes cover exchanges that are registered as in progress before or
> during the scan, but not an exchange that is already in the recovery store
> and not yet registered.
> The pessimistic window can also be reproduced without pausing anything inside
> Camel's own code: with {{completionFromBatchConsumer}} several groups are
> completed by one exchange, and the others wait in the completed store,
> unregistered, while the first one is submitted.
> Proposed fix: in {{onCompletion}}, before calling
> {{aggregationRepository.remove}} on a recoverable repository, mark the
> aggregated exchange as "being completed". {{onSubmitCompletion}} removes the
> mark once it has registered the id in {{inProgressCompleteExchanges}}, and
> the recover task treats a marked id as in progress. The mark is removed again
> when the exchange is not submitted: {{remove}} throws, the exchange is
> discarded, the aggregation fails after a group was completed, or the group is
> force discarded. Those exchanges are then still recovered as before.
> The mark is kept in a separate map with a counter per exchange id instead of
> going straight into {{inProgressCompleteExchanges}}. With optimistic locking,
> two threads can try to complete the same group. The one whose compare-and-set
> remove fails must not clear the registration of the one that succeeded.
> Delivery stays at least once: every marked exchange is either submitted, and
> then covered by the existing in-progress and {{onFailure}} handling, or
> unmarked so the recover task picks it up. The change only prevents the
> recover task from sending an exchange that is about to be sent anyway.
> This does not cover several Camel instances sharing one repository. The
> recover task of another instance can still pick up the exchange; the JDBC
> repository's {{recoveryByInstance}} option exists for that case.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
