[
https://issues.apache.org/jira/browse/CAMEL-25164?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25164:
--------------------------------
Fix Version/s: 4.18.5
Backported to camel-4.18.x via https://github.com/apache/camel/pull/27127
(4.18.5).
_Claude Code on behalf of davsclaus_
> camel-infinispan - the aggregation repositories recover a completed exchange
> without the message that completed the group
> -------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25164
> URL: https://issues.apache.org/jira/browse/CAMEL-25164
> Project: Camel
> Issue Type: Bug
> Components: camel-infinispan
> Reporter: shashank
> Assignee: shashank
> Priority: Minor
> Fix For: 4.18.5, 4.23.0
>
>
> Since CAMEL-24622, {{InfinispanAggregationRepository}} (the shared base of
> {{InfinispanEmbeddedAggregationRepository}} and
> {{InfinispanRemoteAggregationRepository}}) keeps a completed exchange for
> recovery under a {{"camel-recovery:" + exchangeId}} key in the same cache.
> {{remove(ctx, key, exchange)}} stores the entry it removed from the cache
> there, and only uses the exchange passed to {{remove}} when the cache had no
> entry:
> {code:java}
> DefaultExchangeHolder holder = getCache().remove(key);
> if (useRecovery) {
> if (holder == null) {
> holder = DefaultExchangeHolder.marshal(exchange, true,
> allowSerializedHeaders);
> }
> getCache().put(recoveryKey(exchange.getExchangeId()), holder);
> }
> {code}
> When an incoming message completes a group, the Aggregate EIP aggregates it
> into the group and calls {{remove}} without adding the final state to the
> repository first ({{AggregateProcessor}} only calls {{add}} for a group that
> is not complete). The entry in the cache is therefore the group before the
> last message. If the processing after the aggregator fails, the recover task
> sends the stored entry, and the recovered exchange is missing the message
> that completed the group.
> This is the defect CAMEL-24946 fixed for {{KeyValueAggregationRepository}},
> and the Caffeine and Ehcache repositories store the given exchange as well
> (CAMEL-25153). Claus Ibsen pointed out the Infinispan case in the review of
> apache/camel#27108.
> h3. Reproduction
> Route {{from("direct:start").aggregate(header("id"),
> strategy).aggregationRepository(repo).completionSize(3).to("mock:aggregated").process(failOnce)}}
> with an {{InfinispanEmbeddedAggregationRepository}} and
> {{recoveryInterval=100}} (only to make the test fast). The strategy appends
> the body to the old exchange and returns it. After sending "a", "b" and "c"
> for one group, the mock receives "a+b+c", then the recovered exchange "a+b"
> with {{CamelRedelivered=true}}. The message "c" is lost.
> h3. Proposed fix
> In {{remove}}, store the marshalled exchange passed to {{remove}} under the
> recovery key, as {{KeyValueAggregationRepository}} and the Caffeine and
> Ehcache repositories do. The embedded and the remote repository share this
> code, so both are fixed. With the fix the route above receives "a+b+c" twice
> (the second time as redelivered).
> Tests: a route test in camel-infinispan-embedded (the one above) and an
> operations test that the exchange given to {{remove}} is recovered. Both fail
> without the fix ({{expected: <a+b+c> but was: <a+b>}}). The remote repository
> tests are integration tests that need a container.
> Affected: the versions with the CAMEL-24622 recovery store. The same code is
> on main and on the camel-4.18.x and camel-4.22.x branches.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
