[
https://issues.apache.org/jira/browse/CAMEL-24944?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-24944.
---------------------------------
Fix Version/s: 4.23.0
Resolution: Fixed
The fix is merged on main, so it is in Camel 4.23.0:
* c949b429a938 CAMEL-24944: camel-core - Aggregate EIP: send a pre-completed
group also when the next exchange fails to aggregate
Resolving, as the ticket was not updated when the PR was merged.
_Claude Code on behalf of Claus Ibsen_
> Aggregate EIP: with optimistic locking and a pre-completing
> AggregationStrategy, a completed group can be lost
> --------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24944
> URL: https://issues.apache.org/jira/browse/CAMEL-24944
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Priority: Minor
> Fix For: 4.23.0
>
>
> With {{optimisticLocking()}} and an {{AggregationStrategy}} that uses
> pre-completion ({{canPreComplete()}} returns true), a group that was
> pre-completed can be removed from the repository and never sent.
> In pre-completion mode {{doAggregation}} first completes the existing group:
> it removes it from the repository and keeps it in a local list. Then it
> stores the new exchange as the first exchange of a new group with {{add(key,
> null, newExchange)}}. If another thread created a new group for the same key
> in between, that add throws {{OptimisticLockingException}}. The exception
> leaves {{doAggregation}} before the local list is returned, and
> {{doInOptimisticLock}} retries the exchange from the beginning. The group
> that was already removed is never sent.
> Example, key 1, strategy pre-completes when the body starts with {{START}}:
> # {{a1}} is aggregated. The repository holds group [a1].
> # {{START-b}} arrives. Its thread removes [a1] from the repository
> (pre-completion).
> # Before it stores its new group, {{c}} arrives on another thread and creates
> group [c].
> # The add of {{START-b}} fails with {{OptimisticLockingException}} and is
> retried. The retry pre-completes [c] and starts [START-b].
> Observed with {{MemoryAggregationRepository(true)}} and the {{START-b}}
> thread paused after its {{remove()}}; the remaining groups are flushed with
> {{forceCompletionOfAllGroups()}} at the end:
> {noformat}
> downstream output: [c (completedBy=strategy), START-b (completedBy=force)]
> {noformat}
> {{a1}} is never delivered. Without the pause the output is {{[a1
> (completedBy=strategy), START-b+c (completedBy=force)]}}.
> Cause: {{AggregateProcessor.doProcess}} catches only
> {{CamelExchangeException}} around {{doAggregation}}, so the
> {{OptimisticLockingException}} thrown by {{doAggregationRepositoryAdd}} after
> a successful pre-completion discards the list of completed exchanges.
> The same list is dropped in pessimistic mode too, whenever the rest of
> {{doAggregation}} does not finish normally after a pre-completion:
> * the strategy throws when it aggregates the new exchange: the new exchange
> fails as expected, but the pre-completed group is lost as well,
> * the same with {{discardOnAggregationFailure}}: {{doAggregation}} returns
> {{null}} for the discarded first exchange of the new group, and the list with
> the pre-completed group is dropped.
> With a recoverable repository the lost group was eventually re-delivered by
> the recover task (from its completed store). With
> {{MemoryAggregationRepository}} it was lost.
> Proposed fix: {{doAggregation}} adds completed exchanges to a list owned by
> {{doProcess}}, and {{doProcess}} submits that list in a {{finally}} block
> after releasing the lock. So groups that were already completed (and removed
> from the repository) are sent even if the aggregation fails afterwards or the
> exchange is retried because of optimistic locking.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
