shashank created CAMEL-26603:
--------------------------------
Summary: camel-core - Aggregate EIP: forceCompletionOnStop loses
the messages that the consumer routes during the graceful shutdown
Key: CAMEL-26603
URL: https://issues.apache.org/jira/browse/CAMEL-26603
Project: Camel
Issue Type: Bug
Components: camel-core
Reporter: shashank
Assignee: shashank
Found with Fray (controlled concurrency testing for the JVM): two threads
sending to an aggregator with {{forceCompletionOnStop}} while a third stops the
route; every send returned without error, but only the messages that arrived
before the stop came out of the aggregator.
When a route stops, {{DefaultShutdownStrategy}} first calls
{{prepareShutdown(false, false)}} on the route services, so the aggregator
completes its groups, and then waits for the inflight and pending exchanges.
Consumers that defer their shutdown (seda, direct, and others) keep routing
during that wait; seda drains its queue. Their messages start new groups. After
the wait the strategy stops the deferred consumers and calls
{{prepareShutdown}} again, and the aggregator completes the new groups then.
But stopping a {{DefaultConsumer}} stops its processor, which is the whole
route, so the aggregated exchange fails with {{RouteStoppingException}} ("The
exchange cannot continue: its route is being stopped or reloaded"), logged at
WARN by the aggregator, and those messages are lost, although their senders
(and a seda queue) had handed them over.
{code:java}
from("seda:in")
.aggregate(constant(true),
AggregationStrategies.groupedBody()).completionSize(100)
.forceCompletionOnStop()
.to("mock:out");
// one message aggregated, one in flight, two waiting in the seda queue, then
stopRoute:
// mock:out gets only the first group; the other three messages are lost (WARN
with RouteStoppingException)
{code}
The same happens when the CamelContext stops.
Expected: with {{forceCompletionOnStop}}, the messages that the route still
accepts while it stops are completed and sent on too.
Suggested fix: in {{DefaultShutdownStrategy}}, prepare the route services again
(not forced) after the deferred consumers have been prepared and before they
are stopped, so the aggregator completes these groups while the route can still
process them (it waits for its in-progress completions there).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
