CAMEL-24622: camel-infinispan - keep completed exchanges for recovery - #26115
CAMEL-24622: camel-infinispan - keep completed exchanges for recovery#26115oscerd wants to merge 1 commit into
Conversation
The aggregation repository implements RecoverableAggregationRepository but had no recovery store, so the four recovery methods worked on the wrong key space: remove deleted the completed exchange outright, confirm removed an exchange id from a cache keyed by correlation key, scan returned the correlation keys of the aggregations still in progress, and recover therefore returned an exchange that was still aggregating. With useRecovery enabled by default, the recovery task re-delivered aggregations that were still accumulating, marked them CamelRedelivered and sent them to the dead letter channel once maximumRedeliveries was reached, while an exchange that genuinely failed after completion could never be recovered. A completed exchange is now kept in the same cache under a camel-recovery:<exchange id> key until it is confirmed. That is what scan reports and recover reads, confirm deletes it, and getKeys filters those entries out so it still reports only the aggregations in progress. The key follows the exchange handed to remove, as JdbcAggregationRepository does. Recovery entries are only written when useRecovery is enabled. The three existing tests asserted the broken key space and are rewritten around the real contract, in the embedded test and in the remote integration test. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JpshGTyfSmdq7hC8tuEwrZ
gnodet
added
port/camel-4.22.x
🔄 Backport BotThis bugfix targets
Labels Port PRs will be created automatically when this PR is merged. Comment ℹ️ If you push additional commits after |
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 14 tested, 26 compile-only — current: 13 all testedMaveniverse Scalpel detected 40 affected modules (current approach: 13).
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Solid fix — the recovery key space is correctly implemented, tests are thorough, and the upgrade guide is excellent.
The remove → put compound action across cache operations has a theoretical crash window where the exchange could be lost, but this is inherent to the single-cache design (the two-table JDBC approach has the same window between DELETE and INSERT), and the PR description explicitly acknowledges the trade-off and offers to rework with a second cache if preferred.
ast-grep flagged scan() for mutable-collection-return — false positive, the code already wraps with Collections::unmodifiableSet via Collectors.collectingAndThen.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
2 participants
Found by a source audit of
components/camel-infinispan.The problem
InfinispanAggregationRepositoryimplementsRecoverableAggregationRepository, but it keeps a single cachekeyed by the aggregation correlation key and has no recovery store, so all four recovery methods work on the
wrong key space:
remove(ctx, key, exchange)cache.remove(key)— the completed exchange is goneconfirm(ctx, exchangeId)cache.remove(exchangeId)against the correlation-keyed cache — never matchesscan(ctx)recover(ctx, exchangeId)With
useRecoverydefaulting totrue, the consequences are visible at runtime:AggregateProcessortreats thescan()result as exchange ids and checks it againstinProgressCompleteExchanges, so the duplicate-delivery guard never matches;CamelRedeliveredand re-submitted on every recovery interval(default 5s);
maximumRedeliveries(default 3) it goes to the dead letter channel;the interface.
Neither the remote nor the embedded subclass overrides these methods, so both are affected.
The change
A completed exchange is kept in the same cache under a
camel-recovery:<exchange id>key until it isconfirmed.
scan()reports those ids,recover()reads them,confirm()deletes one, andgetKeys()filters them out so it still reports only the aggregations in progress.
scan()returns an empty set whenuseRecovery=false.The recovery entry is keyed by
exchange.getExchangeId()of the exchange handed toremove(), which is whatJdbcAggregationRepositorydoes (final String confirmKey = exchange.getExchangeId()), so the confirm pathlines up with the canonical implementation.
On the single-cache design.
RedisAggregationRepositoryandJdbcAggregationRepositoryuse a secondmap/table for the recovery store, and that was the first thing I tried. For Hot Rod it means the user has to
provision a second cache server-side — a deployment change I do not think belongs in a bug fix — so this
keeps one cache and namespaces the recovery keys instead. If you would rather have the second cache with an
option to name it, say so and I will rework it.
Tests
The three existing tests encoded the broken key space —
testConfirmExistconfirmed with a correlation keywhile the exchanges carried
Exchange_Nids,testScannamed its resultexchangeIdSetand assertedcorrelation keys,
testRecoverrecovered by correlation key. They are rewritten around the real contract(add → remove → scan sees the exchange id → recover returns it → confirm clears it) in both the embedded
test and the remote IT, plus a new test that
getKeys()ignores recovery entries.mvn clean installoncomponents/camel-infinispanis green — 111 remote tests, including the rewrittenInfinispanRemoteAggregationRepositoryOperationsITagainst a Hot Rod testcontainer, and 12 embeddedaggregation tests. Full reactor
mvn clean install -DskipTests -Dquicklygreen. Upgrade-guide entry addedfor 4.23.
Claude Code on behalf of oscerd
🤖 Generated with Claude Code