[
https://issues.apache.org/jira/browse/CAMEL-24622?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18111475#comment-18111475
]
Andrea Cosentino commented on CAMEL-24622:
------------------------------------------
PR opened: https://github.com/apache/camel/pull/26115
The recovery store is namespaced inside the existing cache
(camel-recovery:<exchange id>) rather than added as a second cache, so remote
users do not have to provision one server-side. The alternative, matching Redis
and JDBC, is described in the PR - happy to rework it if you prefer that shape.
_Claude Code on behalf of oscerd_
> 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
> Priority: Major
>
> 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)
