[
https://issues.apache.org/jira/browse/CAMEL-25837?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-25837.
---------------------------------
Resolution: Fixed
> 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
> Fix For: 4.24.0
>
>
> 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. It still waits up to
> {{pollingTimeout}} for an item, but in {{IQueue.poll}} calls of at most one
> second, so it sees a stop within a second (an idle consumer makes one poll
> call per second instead of one per {{pollingTimeout}}); {{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}} (3 tests, an embedded
> Hazelcast member). All three 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>}}),
> {{exchangeInFlightCompletesWhenTheRouteStops}} ({{mock://slow Received
> message count. Expected: <1> but was: <0>}}) and
> {{longPollingTimeoutWaitsForAnItemAndStillStopsPromptly}}
> ({{pollingTimeout=60000}}: an item offered after 3 s of waiting is routed,
> the stop takes less than 5 s, and the item offered afterwards stays in the
> queue; on main that item is taken). All three pass with the fix.
> camel-hazelcast unit tests with the fix: 245 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
(v10.3.26#10030026)
