Andrea Cosentino created CAMEL-24622:
----------------------------------------
Summary: camel-infinispan: the aggregation repository implements
RecoverableAggregationRepository without a recovery store
Key: CAMEL-24622
URL: https://issues.apache.org/jira/browse/CAMEL-24622
Project: Camel
Issue Type: Bug
Components: camel-infinispan
Reporter: Andrea Cosentino
Assignee: Andrea Cosentino
h3. Summary
{{InfinispanAggregationRepository}} implements
{{RecoverableAggregationRepository}}, but it keeps a single cache keyed by the
aggregation correlation key and has no recovery store. The four recovery
methods therefore operate on the wrong key space.
h3. Detail
{{AggregateProcessor}} drives a two-key-space contract: on completion it calls
{{remove(ctx, key, original)}}, which a recoverable repository is expected to
use to move the marshalled exchange into a recovery store keyed by *exchange
id*; {{confirm(ctx, exchangeId)}} then deletes it from there; the recovery task
calls {{scan(ctx)}} expecting a set of *exchange ids* and loads each with
{{recover(ctx, exchangeId)}}.
In {{camel-infinispan-common}} ({{InfinispanAggregationRepository}}):
* line 97 - {{remove(...)}} does a plain {{getCache().remove(key)}}. The
completed exchange is deleted outright, so nothing is ever available to recover.
* line 103 - {{confirm(...)}} does {{getCache().remove(exchangeId)}} against
the correlation-keyed cache. It never matches; on the Hot Rod repository this
is a wasted network round trip per completed aggregation.
* line 114 - {{scan(...)}} returns {{getCache().keySet()}}, i.e. the
correlation keys of the aggregations that are still in flight.
* line 122 - {{recover(...)}} is therefore called with a correlation key and
returns an exchange that is still aggregating.
Consequences, with {{useRecovery}} defaulting to {{true}} (line 47):
* {{AggregateProcessor}} line 1480 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 {{REDELIVERED}} and re-submitted to the
route on every recovery interval (default 5s).
* After {{maximumRedeliveries}} (default 3) it is sent to the dead letter
channel.
* Exchanges that genuinely completed and then failed downstream are never
recoverable, which is the entire point of the interface.
Neither {{InfinispanRemoteAggregationRepository}} nor
{{InfinispanEmbeddedAggregationRepository}} overrides these methods.
h3. Reference implementation
{{RedisAggregationRepository}} shows the correct shape for a key/value store: a
second map ({{persistedCache}}) keyed by exchange id, written in {{remove()}}
(line 310), deleted in {{confirm()}} (line 353), enumerated by {{scan()}} (line
167) and read by {{recover()}} (line 181). {{JdbcAggregationRepository}} does
the same with a completed table.
h3. Note on tests
{{InfinispanEmbeddedAggregationRepositoryOperationsTest}} currently encodes the
wrong key space - it confirms with a correlation key while the exchanges carry
{{Exchange_N}} ids, names the {{scan()}} result {{exchangeIdSet}} but asserts
correlation keys, and recovers by correlation key. Those tests have to be
reworked with the fix.
Found by a source audit of {{components/camel-infinispan}} at 10f750a2d61f.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
