[
https://issues.apache.org/jira/browse/CAMEL-25156?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen reassigned CAMEL-25156:
-----------------------------------
Assignee: shashank
> camel-caffeine - CaffeineIdempotentRepository.add() is a check-then-put, so
> two concurrent exchanges with the same message id both pass the Idempotent
> Consumer
> ---------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25156
> URL: https://issues.apache.org/jira/browse/CAMEL-25156
> Project: Camel
> Issue Type: Bug
> Components: camel-caffeine
> Reporter: shashank
> Assignee: shashank
> Priority: Minor
> Fix For: 4.23.0
>
>
> The Idempotent Consumer EIP (eager, the default) calls
> {{repository.add(key)}} without a lock and processes the message only when it
> returns true, so {{add}} must be atomic. {{MemoryIdempotentRepository}} holds
> a lock around its check and put; the other cache based repositories use an
> atomic {{putIfAbsent}}. {{CaffeineIdempotentRepository.add}} (line numbers of
> main) does not:
> {code:java}
> public boolean add(String key) {
> if (cache.asMap().containsKey(key)) { // :61
> return false;
> } else {
> cache.put(key, true); // :64
> return true;
> }
> }
> {code}
> Two exchanges with the same id that reach the EIP at the same time (a
> consumer with concurrent consumers, such as seda, jms, kafka
> {{consumersCount}}, or a parallel split) can both read "absent" and both
> return true, so the duplicate is processed.
> h3. Reproduction
> * route {{from("direct:in").idempotentConsumer(header("id"),
> repo).process(count)}}, two threads send a message with the same id; the
> repository's cache is replaced (reflection, before start) by a delegating
> Caffeine cache whose {{asMap().containsKey}} waits on a barrier after it read
> the answer, so both calls are between the check and the put: the message is
> processed 2 times. 3 of 3 runs, and 20 of 20 in a loop. Sequential control:
> processed once.
> * no hooks: 8 threads call {{add}} for the same 20000 keys: {{add}} returned
> true more than once for 3536, 1793 and 1338 keys in 3 runs.
> A TLA+ model of two consumers calling {{add}} with the check and the put as
> separate steps violates "the message is not processed by two consumers at the
> same time"; with an atomic add it holds, also with a failed processing that
> removes the key and a redelivery.
> h3. Proposed fix
> {code:java}
> public boolean add(String key) {
> // atomic, so two exchanges with the same key cannot both be added
> return cache.asMap().putIfAbsent(key, Boolean.TRUE) == null;
> }
> {code}
> With the fix the route test processes the message once (3 of 3) and the
> stress test finds no key added twice (3 of 3); the sequential control is
> unchanged and the existing tests of the repository pass
> ({{CaffeineIdempotentRepositoryTest}},
> {{CaffeineIdempotentRepositoryWithSplitTest}}). A deterministic unit test
> (two {{add}} calls held on the key's entry by a {{compute}} of the cache map
> after they checked for the key) fails without the fix (both return true) and
> passes with it; camel-caffeine passes 81 tests.
> Affected: all versions (the same code at camel-3.20.0, 4.0.0, 4.10.0, 4.14.0,
> 4.18.0, 4.22.0 and main).
> Duplicate check (2026-09-30): JIRA text "CaffeineIdempotentRepository"
> (none), component camel-caffeine since 2025 (CAMEL-23411 object filter,
> CAMEL-22759 test context), text "idempotent" with "atomic" (none for
> Caffeine); GitHub pull requests "CaffeineIdempotentRepository", "caffeine
> idempotent": none.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
