CAMEL-24315: camel-google-pubsub - stop subscribers that finish starting after the consumer stopped - #25248
Open
henrik242 wants to merge 1 commit into
Open
Conversation
…ing after the consumer stopped
Contributor
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
gnodet
approved these changes
Jul 30, 2026
gnodet
left a comment
gnodet
left a comment
Contributor
There was a problem hiding this comment.
Correct, minimal fix for a real race condition where subscribers or pull futures starting during consumer stop are missed by doStop(), causing threads to park forever.
Why the fix is correct:
The double-check pattern works because BaseService.stop() sets status = STOPPING (volatile) BEFORE calling doStop(). Combined with the volatile visibility guarantee, either (a) doStop() sees the subscriber in the list and stops it, or (b) the re-check sees !isRunAllowed() and stops it. Both paths cannot be missed simultaneously.
Additional observations:
- Calling
stopAsync()in the re-check AND in thefinallyblock is safe — Guava'sAbstractApiService.stopAsync()is idempotent. - For the synchronous pull path, cancelling the future causes
future.get()to throwCancellationException, which is already caught at line 398 — no unhandled exception risk. - The tests are well-designed: they use
CountDownLatchbarriers to deterministically reproduce the race rather than relying on timing. - The PubSub Lite consumer (also modified in CAMEL-22898) no longer exists in the codebase, so no parallel fix is needed.
- Minor style notes (not blocking): test class uses
publicmodifier and JUnit assertions instead of the project-preferred package-private + AssertJ, but this is consistent with the existingGooglePubsubConsumerShutdownTestin the same package.
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
Claude Code on behalf of Guillaume Nodet
Contributor
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 9 tested, 29 compile-only — current: 9 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 9).
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Since CAMEL-22898 (4.18.0), GooglePubsubConsumer adds a subscriber to the subscribers list only after startAsync().awaitRunning() returns. doStop() and deferShutdown() only stop subscribers already in that list, so stopping the consumer while a subscriber is still starting misses it. Nothing ever calls stopAsync() on it, and the wrapper thread parks forever in awaitTerminated(), an uninterruptible Guava wait.
The consumer executor then never terminates: context shutdown waits 2 x shutdownAwaitTermination (default 10s) per stuck pool and the thread leaks even after forced shutdown. Test suites that stop many CamelContexts (e.g. Spring Boot integration tests) pay up to 20s per leaked consumer per context; we saw a CI build go from 5 to 25 minutes from this alone.
The synchronous pull path has the same window: the pull future is added to pendingSynchronousPullResponses only after futureCall() returns, so a stop landing in between misses it and get() blocks until the pull deadline instead of being cancelled.
Fix: after registering the subscriber or pull future, re-check the consumer state and stop/cancel it if the consumer is no longer allowed to run. stop() sets the volatile service status to STOPPING before doStop() runs, so either doStop() sees the subscriber in the list or the re-check sees the stopped status. Unit tests reproduce both races deterministically and fail without the fix.
Workaround on affected versions: lower shutdownAwaitTermination to bound the shutdown delay; the thread still leaks.