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