[
https://issues.apache.org/jira/browse/CAMEL-25837?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18128956#comment-18128956
]
shashank commented on CAMEL-25837:
----------------------------------
Correction to the description: the fix does not stop the processor "as the
hazelcast-seda consumer does". The hazelcast-seda consumer does not wait for
its loops on stop either, and loses an item the same way; that is filed
separately.
_Claude Code on behalf of allthingssecurity_
> camel-hazelcast - the queue consumer in poll mode loses items when its route
> stops: the next item offered after the stop, and the item of the exchange in
> flight
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25837
> URL: https://issues.apache.org/jira/browse/CAMEL-25837
> Project: Camel
> Issue Type: Bug
> Components: camel-hazelcast
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> With {{queueConsumerMode=Poll}} the hazelcast-queue consumer
> ({{HazelcastQueueConsumer}}, main 20cdb309f7d0) takes items with
> {{IQueue.poll(pollingTimeout)}} in its own thread, and a poll removes the
> item from the queue. {{doStop}} stopped that thread with {{shutdownNow}},
> which interrupts it. Two things then lose items:
> * *An idle route.* The thread usually waits in a poll. The interrupt ends the
> poll on the caller's side only: the poll operation stays parked on the member
> that owns the queue until its timeout ({{pollingTimeout}}, 10 seconds by
> default). The first item offered in that time is taken by that operation and
> its response goes nowhere: the item is gone from the queue and nobody routes
> it. This happens at every stop of the route: a route stop or restart, a
> {{master:}} or route policy failover, the shutdown of the application, while
> the producers keep offering.
> * *An exchange in flight.* {{doStop}} also stopped the processor of the route
> ({{super.doStop()}}) before the thread was done, and the interrupt ends any
> blocking call of the route. The exchange fails ({{RouteStoppingException}} or
> an {{InterruptedException}}); the consumer does not look at the exchange
> exception, so nothing is logged, and the item it polled is lost. A graceful
> shutdown does not help, since the consumer is stopped first.
> The listen mode (the default) is not affected.
> *Fix*: the consumer lets its thread end on its own. A poll waits at most one
> second (a poll that returns nothing just polls again, so {{pollingTimeout}}
> still works as before from the route's side); {{doStop}} no longer interrupts
> the thread but waits for its current poll and exchange (a graceful executor
> shutdown, bounded by the context's {{shutdownAwaitTermination}}), and stops
> the processor afterwards, as the hazelcast-seda consumer does. A forced
> shutdown still interrupts.
> *Tests*: new {{HazelcastQueueConsumerPollStopTest}} (2 tests, an embedded
> Hazelcast member). Both fail on main:
> {{itemOfferedAfterTheRouteStoppedStaysInTheQueue}} (stop the idle route,
> offer an item: {{The item offered after the route stopped was taken from the
> queue ==> expected: <1> but was: <0>}}) and
> {{exchangeInFlightCompletesWhenTheRouteStops}} ({{mock://slow Received
> message count. Expected: <1> but was: <0>}}). Both pass with the fix.
> camel-hazelcast unit tests with the fix: 244 run, 0 failures, 1 error
> ({{HazelcastUserConfigSerializationFilterWarningTest.warnsForUserClientConfigWithoutFilter}},
> a client that cannot reach a cluster in the test environment; it errors the
> same way without the change).
> *How it was found*: variant analysis of CAMEL-25790 (pattern: a consumer with
> its own loop and executor, and what its stop does to that loop), then the
> JUnit test. The parked poll was not expected from reading the code; the test
> showed it.
> *Affected*: main, camel-4.18.x and camel-4.14.x ({{shutdownNow}} in
> {{doStop}} and {{super.doStop()}} before it; read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
