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