shashank created CAMEL-24944:
--------------------------------
Summary: 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
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)
