[
https://issues.apache.org/jira/browse/CAMEL-25216?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen reassigned CAMEL-25216:
-----------------------------------
Assignee: shashank
> camel-redis - RedisAggregationRepository recovers a completed exchange
> without the message that completed the group
> -------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25216
> URL: https://issues.apache.org/jira/browse/CAMEL-25216
> Project: Camel
> Issue Type: Bug
> Reporter: shashank
> Assignee: shashank
> Priority: Minor
>
> Without optimistic locking and with recovery enabled (the defaults),
> {{RedisAggregationRepository.remove(ctx, key, exchange)}} (line numbers of
> main, :324-327) removes the group in a Redisson transaction and puts the
> entry it removed into the completed map:
> {code:java}
> DefaultExchangeHolder removedHolder = tCache.remove(key);
> tPersistentCache.put(exchange.getExchangeId(), removedHolder);
> {code}
> When an incoming message completes a group ({{completionSize}},
> {{completionPredicate}}), 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 map is therefore the group before the last message. If the
> processing after the aggregator fails, the recover task sends that entry, and
> the recovered exchange is missing the message that completed the group.
> CAMEL-24946 fixed the same defect in {{KeyValueAggregationRepository}} and
> lists this branch of {{RedisAggregationRepository.remove}} as a follow-up
> that was not changed there because its tests need Redis. The optimistic
> branch of the same method already stores the given exchange.
> h3. Reproduction
> Against a local Redis server (8.6, no persistence):
> * route {{from("direct:start").aggregate(header("id"),
> strategy).aggregationRepository(repo).completionSize(3).to("mock:aggregated").process(failOnce)}},
> {{recoveryInterval=100}} (only to make the test fast); the strategy appends
> the body to the old exchange and returns it. After "a", "b", "c" the mock
> receives "a+b+c", then the recovered exchange "a+b"
> ({{CamelRedelivered=true}}): the message "c" is lost. 3 of 3 runs.
> * repository level: {{add}} a group "a+b", read it back, set its body to
> "a+b+c" (what the aggregator does with the message that completes the group),
> {{remove}} it: {{recover}} returns "a+b". 3 of 3 runs.
> h3. Proposed fix
> Put the holder that {{remove}} already marshals from the given exchange at
> the top of the method into the completed map, instead of {{removedHolder}},
> as CAMEL-24946 proposed.
> With the fix both checks recover "a+b+c". Tests: a route IT
> ({{AggregateRedisRecoverIT}}) and an operations IT method
> ({{testPessimisticRemoveRecoversTheGivenExchange}}) in the style of the
> module's ITs, which use the Redis test-infra service. As Docker is not
> available here, they were run as copies pointed at the local Redis server,
> together with the existing {{RedisAggregationRepositoryOperationsIT}} and
> {{AggregateRedisIT}} (15 tests pass); without the fix the two new tests fail
> as above.
> Affected: all versions (the same code at camel-3.20.0, 4.0.0, 4.10.0, 4.14.0,
> 4.18.0, 4.22.0 and main).
> Duplicate check (2026-09-30): JIRA text "RedisAggregationRepository"
> (CAMEL-24946, where this is a named follow-up, CAMEL-24622, CAMEL-23714
> polish); GitHub pull requests "RedisAggregationRepository", "redis
> aggregation recover": none for this defect (#26791 changed only
> KeyValueAggregationRepository). No open pull request touches the file.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
