[jira] [Created] (CAMEL-24943) Aggregate EIP: the recover task can re-deliver an aggregated exchange that is being completed normally, which sends it twice

["shashank (Jira)" <ji...@apache.org>]

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)