oscerd opened a new pull request, #26115:
URL: https://github.com/apache/camel/pull/26115
Found by a source audit of `components/camel-infinispan`.
### The problem
`InfinispanAggregationRepository` implements
`RecoverableAggregationRepository`, but it keeps a single cache
keyed by the aggregation correlation key and has no recovery store, so all
four recovery methods work on the
wrong key space:
| method | contract | what it did |
|---|---|---|
| `remove(ctx, key, exchange)` | move the holder into a recovery store keyed
by **exchange id** | plain `cache.remove(key)` — the completed exchange is gone
|
| `confirm(ctx, exchangeId)` | delete it from the recovery store |
`cache.remove(exchangeId)` against the correlation-keyed cache — never matches |
| `scan(ctx)` | return **exchange ids** to recover | returns the correlation
keys of aggregations still in progress |
| `recover(ctx, exchangeId)` | load a completed, unconfirmed exchange |
returns an exchange that is still aggregating |
With `useRecovery` defaulting to `true`, the consequences are visible at
runtime:
* `AggregateProcessor` treats the `scan()` result as exchange ids and checks
it against
`inProgressCompleteExchanges`, so the duplicate-delivery guard never
matches;
* the still-open aggregation is marked `CamelRedelivered` and re-submitted
on every recovery interval
(default 5s);
* after `maximumRedeliveries` (default 3) it goes to the dead letter channel;
* and an exchange that genuinely failed after completion can never be
recovered, which is the whole point of
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 is
confirmed. `scan()` reports those ids, `recover()` reads them, `confirm()`
deletes one, and `getKeys()`
filters them out so it still reports only the aggregations in progress.
`scan()` returns an empty set when
`useRecovery=false`.
The recovery entry is keyed by `exchange.getExchangeId()` of the exchange
handed to `remove()`, which is what
`JdbcAggregationRepository` does (`final String confirmKey =
exchange.getExchangeId()`), so the confirm path
lines up with the canonical implementation.
**On the single-cache design.** `RedisAggregationRepository` and
`JdbcAggregationRepository` use a second
map/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 — `testConfirmExist`
confirmed with a correlation key
while the exchanges carried `Exchange_N` ids, `testScan` named its result
`exchangeIdSet` and asserted
correlation keys, `testRecover` recovered 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 install` on `components/camel-infinispan` is green — 111 remote
tests, including the rewritten
`InfinispanRemoteAggregationRepositoryOperationsIT` against a Hot Rod
testcontainer, and 12 embedded
aggregation tests. Full reactor `mvn clean install -DskipTests -Dquickly`
green. Upgrade-guide entry added
for 4.23.
---
_Claude Code on behalf of oscerd_
🤖 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]
