[
https://issues.apache.org/jira/browse/CAMEL-25864?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131724#comment-18131724
]
Claus Ibsen commented on CAMEL-25864:
-------------------------------------
Fixed by https://github.com/apache/camel/pull/28383, merged to main for 4.24.0.
_Claude Code on behalf of Claus Ibsen (davsclaus)_
> camel-core - Aggregate EIP: with an auto-discovered KeyValueRepository, the
> recover task of one aggregator sends the exchanges that another aggregator
> completed again, on its own route
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25864
> URL: https://issues.apache.org/jira/browse/CAMEL-25864
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> An Aggregate EIP without an {{aggregationRepository}} wraps the single
> {{KeyValueRepository}} of the registry in a {{KeyValueAggregationRepository}}
> (CAMEL-24463; the "Single backend, multiple patterns" section of the
> KeyValueRepository page says no explicit wiring is needed). Every aggregator
> of the application does this with the same bean. Since CAMEL-25819 (main,
> 4.24.0) the wrapper is set with {{setAggregationRepository}}, so each of
> these aggregators now schedules a recover task.
> {{KeyValueAggregationRepository}} keeps a completed exchange, until it is
> confirmed, under {{completed:<exchangeId>}}, and {{scan}} returns every key
> with that prefix. So the recover task of aggregator B scans the completed
> exchanges of aggregator A as well. {{AggregateProcessor.RecoverTask}} only
> skips the exchange ids in its own {{inProgressCompleteExchanges}} /
> {{unconfirmedCompleteExchanges}}, so B recovers the exchange that A completed
> and is still processing, marks it redelivered and sends it on B's route
> ({{onSubmitCompletion}}), then confirms it. The exchange is delivered twice,
> once on a route that has nothing to do with it. The same happens to an
> exchange that failed on A's route: it is recovered by A and by B. The recover
> tasks run 1 s after start and then every 5 s (the default
> {{recoveryInterval}}, which cannot be changed for an auto-discovered
> repository), so under load every scan picks up the exchanges the other
> aggregators have in flight.
> {code:java}
> // one KeyValueRepository bean in the registry, e.g. a
> MemoryKeyValueRepository or a RedisKeyValueRepository
> from("direct:orders").routeId("orders")
> .aggregate(header("id"), new
> BodyInAggregatingStrategy()).completionSize(2)
> .process(e -> slowCall()) // the aggregated exchange is
> completed, not yet confirmed
> .to("mock:orders");
> from("direct:invoices").routeId("invoices")
> .aggregate(header("id"), new
> BodyInAggregatingStrategy()).completionSize(2)
> .to("mock:invoices"); // receives the "A+B" of the orders
> route
> {code}
> *Fix*: {{KeyValueAggregationRepository}} gets a constructor with a recovery
> name: the completed exchanges are then kept, scanned, recovered and confirmed
> under {{completed[<name>]:<exchangeId>}}, which an adapter without a name
> (prefix {{completed:}}) does not scan. {{AggregateProcessor}} names the
> adapter it creates for an auto-discovered repository after its route id and
> id ({{<routeId>/<id>}}). The open groups keep their keys
> ({{aggregate:<correlationKey>}}), and so does an adapter created without a
> name. No new bean property, so no generated metadata changes. Completed
> entries written under {{completed:}} by an auto-discovered adapter before
> (4.23.x, where they were never recovered, see CAMEL-25819) are not picked up
> by the named adapters.
> *Tests*: new {{AggregateAutoDiscoveredKeyValueRepositorySharedTest}}
> (camel-core): two routes with an aggregator and one
> {{MemoryKeyValueRepository}} bean; the orders route completes a group and
> holds the aggregated exchange downstream across the recover scans. On main it
> fails with {{mock://invoices Received message count. Expected: <0> but was:
> <1>}}; with the fix it passes, the orders route gets "A+B" once and the
> repository is empty at the end.
> {{KeyValueAggregationRepositoryTest.testRecoveryNameKeepsCompletedExchangesApart}}
> covers the adapter. camel-core aggregator and support tests
> ({{org.apache.camel.processor.aggregator.**}},
> {{org.apache.camel.support.**}}, ...): 707 run, 0 failures, 3 skipped;
> camel-support: 136 run, 0 failures.
> {{AggregateAutoDiscoveredKeyValueRepositoryRecoverTest}} (CAMEL-25819) still
> passes.
> *How it was found*: review of the changes merged to main in the last week
> (CAMEL-25819), looking at what the change made reachable: the recover task
> now runs for a repository that every aggregator of the application shares.
> The JUnit test above is the evidence.
> *Affected*: main only (read, not run). On 4.23.x the auto-discovered adapter
> has no recover task (CAMEL-25819 is fixed in 4.24.0 only), so the duplicates
> do not happen there. An application that explicitly wraps the same
> {{KeyValueRepository}} in a {{KeyValueAggregationRepository}} for several
> aggregators has the same overlap on every version with the adapter; the new
> constructor lets it name each adapter.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
