[
https://issues.apache.org/jira/browse/CAMEL-25790?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25790:
--------------------------------
Fix Version/s: 4.24.0
> camel-google-pubsub - the consumer subscribes or pulls again during a
> graceful shutdown (deferShutdown), so the shutdown waits for its timeout, and
> a route restart leaves the previous subscriber loops running
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25790
> URL: https://issues.apache.org/jira/browse/CAMEL-25790
> Project: Camel
> Issue Type: Bug
> Components: camel-google-pubsub
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.24.0
>
>
> {{GooglePubsubConsumer}} (main 7aae27315946) runs one loop per concurrent
> consumer ({{SubscriberWrapper}}): the asynchronous loop starts a
> {{Subscriber}} and waits until it terminates, the synchronous loop pulls,
> routes the messages and pulls again. Both loops continue while
> {{isRunAllowed() && !isSuspendingOrSuspended()}} (lines 253 and 323). That
> check misses two cases.
> *1. Graceful shutdown.* Since CAMEL-23291 the consumer is {{ShutdownAware}}:
> {{deferShutdown}} stops the subscribers ({{stopAsync}}) and cancels the
> pending synchronous pulls, and the shutdown strategy then waits for the
> pending and inflight exchanges before it stops the consumer. The consumer is
> still started during that wait, so the loop sees {{isRunAllowed() == true}}
> and at once starts a new subscriber (asynchronous pull) or issues a new pull
> (synchronous pull, as soon as the cancelled {{get()}} returns). New messages
> keep being routed while the strategy waits for the consumer to drain. With a
> steady flow of messages the pending count does not reach zero: the graceful
> shutdown (route stop, context stop) waits for its full timeout (45 s by
> default; longer than the 30 s a Kubernetes pod gets by default) and then
> forces the shutdown, with exchanges still in flight; their messages are
> delivered again (duplicates). CAMEL-23291 added {{deferShutdown}} to avoid
> exactly that.
> *2. Route restart.* {{doStop}} only calls {{stopAsync}} and returns; the
> subscriber terminates later (streams closed, acks flushed). If the consumer
> is started again before that ({{restartRoute}}, a route controller or a
> reload that stops and starts the route), the old loop wakes up in a started
> consumer and subscribes again, next to the loops of the new start. Each
> restart adds {{concurrentConsumers}} subscriber loops (each with its streams
> and flow-control budget), and the executors of the previous start are only
> shut down in {{doShutdown}}. The loops stop only when the route stops for
> good.
> No message is lost; the effects are a shutdown that does not drain (hang
> until the timeout, forced shutdown, redelivered messages) and subscriber
> loops and threads that grow with every restart.
> *Fix*: each loop gets a generation; the consumer increments it when it
> starts, in {{deferShutdown}} and in {{doStop}}, and a loop of an older
> generation ends instead of subscribing or pulling again (the check after the
> subscriber start and after the pull is issued uses the same condition). A
> start after a stop also shuts down (not gracefully,
> {{ExecutorServiceManager.shutdown}}) the two executors of the previous start;
> their loops end on their own. {{doShutdown}} is unchanged.
> *Tests*: new {{GooglePubsubConsumerStopPullingTest}} (3 tests, mocked
> {{Subscriber}} / {{SubscriberStub}} as in
> {{GooglePubsubConsumerStopRaceTest}}). On main all three fail:
> {{asyncPullDoesNotSubscribeAgainAfterDeferShutdown}} ({{subscribers created
> after the consumer stopped pulling ==> expected: <1> but was: <2>}}),
> {{syncPullDoesNotPullAgainAfterDeferShutdown}} ({{pulls issued after the
> consumer stopped pulling ==> expected: <1> but was: <2>}}),
> {{asyncPullLoopOfPreviousStartEndsWhenConsumerIsRestarted}} ({{subscribers
> created by a start and a restart with one concurrent consumer ==> expected:
> <2> but was: <3>}}). With the fix they pass; camel-google-pubsub unit tests:
> 20 run, 0 failures, 10 skipped (need credentials or the emulator).
> *How it was found*: mutation testing (PIT) showed the consumer's stop /
> shutdown paths and the asynchronous pull ({{CamelMessageReceiver}}, the
> subscriber loop) untested, then reading the code against the
> {{ShutdownAware}} contract (CAMEL-23291: "deferShutdown() should stop the
> message pump while keeping connections alive") and the shutdown strategy
> ({{DefaultShutdownStrategy}}: a deferred consumer stays started until the
> pending exchanges are done), then the JUnit tests.
> *Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (since
> CAMEL-23291, 4.18.3 / 4.21.0: same loop condition with {{deferShutdown}} and
> the executor shut down only in {{doShutdown}}). camel-4.14.x has no
> {{deferShutdown}} and shuts the executor down gracefully in {{doStop}}, which
> waits for the loops to end (read, not run).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
