[
https://issues.apache.org/jira/browse/CAMEL-25837?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25837:
-----------------------------
Description:
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._
was:
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._
> 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. 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
(v8.20.10#820010)
