[jira] [Resolved] (CAMEL-25790) 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

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25790?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen resolved CAMEL-25790.
---------------------------------
    Resolution: Fixed

> 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)