shashank created CAMEL-24949:
--------------------------------
Summary: camel-seda - suspending a route with
purgeWhenStopping=true discards all queued messages
Key: CAMEL-24949
URL: https://issues.apache.org/jira/browse/CAMEL-24949
Project: Camel
Issue Type: Bug
Components: camel-seda
Reporter: shashank
{{purgeWhenStopping}} is documented as "Whether to purge the task queue when
stopping the consumer/route". Suspending the route purges the queue as well:
* {{DefaultShutdownStrategy}} uses the same {{ShutdownTask}} wait loop for
suspend and shutdown.
* The loop asks every {{ShutdownAware}} consumer for
{{getPendingExchangesSize()}} (DefaultShutdownStrategy.java:674, :772-788).
* {{SedaConsumer.getPendingExchangesSize()}} purges the queue whenever
{{purgeWhenStopping}} is set (SedaConsumer.java:88-95). It does not know that
this is only a suspend.
This affects {{routeController.suspendRoute(id)}}, {{camelContext.suspend()}}
and suspending via JMX.
Reproduced against 4.23.0-SNAPSHOT. Route
{{from("seda:work?purgeWhenStopping=...")}}, with 5 messages sent, m1 in
progress and 4 queued, then {{suspendRoute("work")}} and later
{{resumeRoute("work")}}:
{noformat}
purgeWhenStopping=false: after suspendRoute delivered [m1, m2, m3, m4, m5] -> 0
lost
purgeWhenStopping=true : after suspendRoute delivered [m1]; after resumeRoute
still [m1] -> 4 lost
{noformat}
Nothing is logged above DEBUG.
{{SedaConsumer.prepareShutdown(suspendOnly, ...)}} already skips its shutdown
logic when only suspending, so a suspend is clearly not meant to behave like a
stop.
Proposed fix: only purge when the route/consumer is being stopped. The consumer
cannot tell a suspend from a stop while the wait loop runs (the route and
consumer are still Started), so the shutdown strategy has to pass it:
* add a default method {{ShutdownAware.getPendingExchangesSize(boolean
suspendOnly)}} (camel-api, {{@since 4.23}}) that delegates to
{{getPendingExchangesSize()}};
* {{DefaultShutdownStrategy}} calls it with its {{suspendOnly}} flag (the
protected static {{getPendingInflightExchanges(order)}} is kept and delegates
with {{false}});
* {{SedaConsumer}} overrides it and purges only when {{!suspendOnly}}. A
suspend then completes the queued messages first, as it already does without
{{purgeWhenStopping}}. The purge on stop (loop and {{doStop()}}) is unchanged.
The alternative of calling {{prepareShutdown(true, false)}} on deferred
consumers before the loop was not taken: it changes the call sequence for every
{{ShutdownAware}} consumer. The SPI addition is the one design point for
reviewers.
A PR with the fix follows, with regression test
{{SedaPurgeWhenStoppingSuspendTest}}, plus an upgrade-guide note.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
