shashank created CAMEL-25864:
--------------------------------
Summary: 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
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
(v8.20.10#820010)
