shashank created CAMEL-26641:
--------------------------------
Summary: camel-core - Aggregate EIP with forceCompletionOnStop or
completeAllOnStop completes the groups when its route is only suspended
Key: CAMEL-26641
URL: https://issues.apache.org/jira/browse/CAMEL-26641
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
Assignee: shashank
Found while checking the shutdown hooks of the EIPs ({{ShutdownPrepared}} /
{{ShutdownAware}}) for state that a suspend should not touch, after CAMEL-26633
(Loop EIP). No race is needed, the order alone is enough.
The shutdown strategy calls {{prepareShutdown}} on the route services also when
a route is only suspended ({{suspendOnly=true}}).
{{AggregateProcessor.prepareShutdown}} ignores that flag, so with
{{forceCompletionOnStop}} or {{completeAllOnStop}} a route suspend completes
every group and sends the partial aggregations on. After the resume, the rest
of each group starts a new group:
{code:java}
from("direct:start").routeId("agg")
.aggregate(header("id"), strategy).completionSize(3).forceCompletionOnStop()
.to("mock:result");
template.sendBodyAndHeader("direct:start", "A", "id", "foo");
template.sendBodyAndHeader("direct:start", "B", "id", "foo");
controller.suspendRoute("agg"); // mock:result gets "A+B"
controller.resumeRoute("agg");
template.sendBodyAndHeader("direct:start", "C", "id", "foo"); // "C" starts a
new group, "A+B+C" is never sent
{code}
A CamelContext suspend does not complete the groups
({{forceCompletionOfAllGroups}} refuses while the context is not started, and
logs a WARN on every suspend). But with {{completeAllOnStop}},
{{getPendingExchangesSize()}} still reports every group as pending, so
{{camelContext.suspend()}} waits for the whole shutdown timeout and then forces
the routes down (shutdownNowOnTimeout).
Ways to get there: a route suspend from JMX, the control bus, the CLI,
{{DurationRoutePolicy}} or the supervising route controller, and a CamelContext
suspend. Route policies that only suspend the consumer
({{ThrottlingInflightRoutePolicy}}, the scheduled route policies) do not go
through the shutdown strategy and never completed the groups.
The docs of both options talk only about a stop. The error handlers
({{RedeliveryErrorHandler}}) and the seda consumer (CAMEL-24949) already ignore
a suspend in {{prepareShutdown}} / {{getPendingExchangesSize(boolean
suspendOnly)}}.
Expected: a suspend keeps the groups (they go on aggregating after the resume,
{{completionTimeout}} and {{completionInterval}} still apply), and a stop
completes them.
Suggested fix: ignore a suspend in {{AggregateProcessor.prepareShutdown}}, and
return 0 from {{getPendingExchangesSize(boolean suspendOnly)}} for a suspend.
Main only ({{ShutdownAware.getPendingExchangesSize(boolean)}} is new in 4.23),
with an upgrade guide note.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
