shashank created CAMEL-25790:
--------------------------------
Summary: 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
{{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)
