allthingssecurity opened a new pull request, #29407: URL: https://github.com/apache/camel/pull/29407
# Description [CAMEL-26711](https://issues.apache.org/jira/browse/CAMEL-26711) Since CAMEL-23996 (#24691), `KafkaIdempotentRepository.add` puts the key in the local cache (`putIfAbsent(key, key)`), sends the add record synchronously, and rolls the cache back with `cache.remove(key, key)` when the send fails. The poller thread applies the records of the other instances with `cache.put(messageId, messageId)`. If it applies the add of the same key by a peer while this instance's send is in flight, and the send then fails, the rollback cannot tell the two apart and removes the peer's key too. The key is then missing locally although a peer added and processed it, so the next copy of the message is processed again. Change: - `add()` stores a value of its own (`new Object()`) and the rollback removes only that value: `cache.remove(key, token)`. The poller's `put(key, key)` replaces it, so a peer's add applied during the send survives the rollback. `KafkaKeyValueRepository.putIfAbsent` already rolls back this way (`cache.remove(key, newEntry)`). - `remove()` uses `cache.remove(key)`, as the value may now be a token. On main every value equals its key, so this is the same behaviour. - `contains()` (`containsKey`) and `confirm()` (no-op) are unchanged. Nothing reads the cache values. Not changed here (separate issues, can be follow-ups): a failed `remove` broadcast is not compensated; an add that reached the topic although the send reported a failure is still rolled back; in non-eager mode the rollback forgets a key that was already processed; `KafkaKeyValueRepository.put()`/`delete()` roll back unconditionally. This fix holds whatever is decided for those. CAMEL-23996 is also on camel-4.22.x, where the same rollback exists; that branch predates the `KafkaChangelog` refactoring, so the fix would need a manual port. Test (`KafkaIdempotentRepositoryTest`, no broker, `MockProducer` that does not complete on its own): a thread calls `add("key1")` and waits in the send; the test waits until the record is in `history()`, applies a peer's `add key1` through the poller method `addToCache`, then fails the send with `errorNext`. The key must still be present, and a second `add("key1")` must return false. On main the new test fails (the key is gone); with the fix 10/10. Mutant (token without the `remove()` change): the 2 existing remove tests fail. 20 runs: 200/200. The same test with the real `SimpleLRUCache` instead of a `ConcurrentHashMap`: passes with the fix, fails on main. camel-kafka unit tests: 308 run, 0 failures, 1 skipped (offline, `-DskipITs`, with the formatter and import-sort plugins). # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-26711](https://issues.apache.org/jira/browse/CAMEL-26711). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built the core modules and camel-kafka with the formatter and import-sort plugins and ran the camel-kafka unit tests; no generated file changes. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
