shashank created CAMEL-24945:
--------------------------------
Summary: Aggregate EIP: with optimistic locking a group can lose
its completion timeout and never complete
Key: CAMEL-24945
URL: https://issues.apache.org/jira/browse/CAMEL-24945
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
With {{optimisticLocking()}} and {{completionTimeout}}, a new group can be left
in the repository without a timeout. It then only completes if another exchange
arrives for the same key, or when the route is stopped with
{{forceCompletionOnStop}}. With an in-memory repository it is lost when the
route stops.
The timeout map is keyed by correlation key. When a group is completed (for
example by {{completionSize}}), {{onCompletion}} first removes the group from
the repository and then calls {{timeoutMap.remove(key)}}. With optimistic
locking there is no lock around these two calls. If another exchange for the
same key starts a new group in between, it registers its timeout
({{trackTimeout}} runs before the repository add) and adds the group. The
completing thread then removes that new timeout entry.
Example with {{completionSize(2).completionTimeout(500)}}, key 1:
# {{a}} is aggregated.
# {{b}} completes [a, b] by size. Its thread removes the group from the
repository.
# {{c}} arrives on another thread, registers a timeout for key 1 and adds group
[c].
# The thread of {{b}} calls {{timeoutMap.remove("1")}}, which removes the
timeout of [c].
Observed with {{MemoryAggregationRepository(true)}} and the thread of {{b}}
paused after its {{remove()}}, waiting 2.5 seconds afterwards:
{noformat}
repo before final flush={1=c}
downstream output: [a+b (completedBy=size)]
{noformat}
Without the pause: {{[a+b (completedBy=size), c (completedBy=timeout)]}}.
In pessimistic mode both calls are made under the aggregation lock, so this
only affects optimistic locking.
Proposed fix: with optimistic locking, do not remove the timeout entry in
{{onCompletion}}. A leftover entry is harmless because the timeout eviction
reads the group and removes it with a compare-and-set, and every new group
refreshes the entry before it is added. In the same mode, {{onEviction}} should
not skip an eviction because the exchange id in the entry is in progress: with
leftover entries that id can belong to a completed group while a newer group
for the key is waiting. The compare-and-set already prevents completing a group
twice. We checked this change in a TLA+ model. Two other options still leave a
group without a timeout: removing the entry only when it holds an exchange id
of the completed group, and registering the timeout after the add. The second
part is needed too: the model, and a unit test with a repository that keeps
exchange ids (like JDBC), strand a group when only the first part is applied.
For example, m1 and m2 start a new group at the same time, and m2's timeout
entry overwrites m1's. m3 then completes [m2, m3], and m1's add succeeds
afterwards. The entry for the key still holds m2's id, which is in progress, so
the eviction is skipped and [m1] never times out.
Side effects, with optimistic locking only: after a group completes, its
timeout entry stays until it expires. The eviction then does one repository
{{get}} for the key, which returns nothing or a newer group. A newer group
created on the same Camel instance always refreshes the entry. With several
instances sharing the repository, an old entry on one instance can complete a
group that another instance created for the same key within
{{completionTimeout}}, so it completes up to one timeout early. Cross-instance
timeouts already work this way for exchanges added to an existing group from
another instance.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
