[
https://issues.apache.org/jira/browse/CAMEL-24941?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-24941.
---------------------------------
Fix Version/s: 4.23.0
Resolution: Fixed
The fix is merged on main, so it is in Camel 4.23.0:
* 8e781863ab13 CAMEL-24941: camel-core - Aggregate EIP: check the closed
correlation key again under the lock
* 4cde7ce99276 CAMEL-24941: camel-core - Make the new test class package-private
* 9d8866747c35 CAMEL-24941: camel-core - Assert that the producer threads of
the new test completed
Resolving, as the ticket was not updated when the PR was merged.
_Claude Code on behalf of Claus Ibsen_
> Aggregate EIP: with closeCorrelationKeyOnCompletion an exchange that arrives
> while its group is being completed starts a new group for the closed key
> -----------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-24941
> URL: https://issues.apache.org/jira/browse/CAMEL-24941
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Priority: Minor
> Fix For: 4.23.0
>
>
> {{closeCorrelationKeyOnCompletion}} is documented to reject exchanges whose
> correlation key has already completed with a
> {{ClosedCorrelationKeyException}}. An exchange that arrives while the group
> for its key is being completed can get past that check and is then aggregated
> into a new group for the closed key. That group is completed again later (by
> size or timeout), so the key produces two aggregated exchanges.
> Example with
> {{aggregate(header("id")).completionSize(2).completionTimeout(500).closeCorrelationKeyOnCompletion(100)}}
> and messages A, B, C for key 1:
> # A is aggregated.
> # B arrives and completes the group (size 2). While B's thread holds the
> aggregation lock, C arrives on another thread, passes the closed-key check
> (the key is not closed yet) and waits for the lock.
> # B completes the group and closes key 1.
> # C gets the lock and is aggregated into a new group for key 1. It is not
> rejected.
> # The new group completes by timeout.
> Observed with a strategy that pauses B inside {{aggregate()}} until C is
> waiting for the lock:
> {noformat}
> C exception: null (expected ClosedCorrelationKeyException)
> D exception: ClosedCorrelationKeyException: The correlation key [1] has been
> closed.
> aggregated output for key 1: [A+B (completedBy=size), C (completedBy=timeout)]
> {noformat}
> Cause: {{AggregateProcessor.doProcess(Exchange, AsyncCallback)}} checks
> {{closedCorrelationKeys}} before taking the aggregation lock, and
> {{doAggregation}} does not check again. With optimistic locking the check
> also runs only once, before the retry loop.
> Proposed fix: check {{closedCorrelationKeys}} again inside the lock (at the
> start of {{doAggregation}}), and before each optimistic-locking retry, and
> fail the exchange with {{ClosedCorrelationKeyException}} if the key has been
> closed in the meantime.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
