Stop consuming once the shutdown has started - #2834
Open
delthas wants to merge 1 commit into
Open
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files
... and 2 files with indirect coverage changes
@@ Coverage Diff @@
## improvement/BB-833/leave-group-on-shutdown #2834 +/- ##
==============================================================================
+ Coverage 75.57% 75.84% +0.26%
==============================================================================
Files 200 200
Lines 13974 13976 +2
==============================================================================
+ Hits 10561 10600 +39
+ Misses 3403 3366 -37
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
close() drains the in-flight work before releasing the partitions, but nothing stopped the fetch loop while it waited: every completed task re-armed _tryConsume(), so the pipeline refilled as fast as it drained and the departure was delayed by work that arrived after the shutdown had begun. Measured against a 3000 message backlog, close() took 6.2s and started 301 further tasks at concurrency 10, and 9.5s and 1864 further tasks with shorter ones; with the guard both are 0 further tasks, in 175ms and 31ms. The same guard ends the self-rescheduling consume loop, which otherwise kept polling a closed client for the lifetime of the process. Issue: BB-833
delthas
force-pushed
the
improvement/BB-833/stop-consuming-on-shutdown
branch
from
September 1, 2026 09:15
a100b8f to
653f932
Compare
delthas
marked this pull request as ready for review
September 1, 2026 09:35
| // only delays the departure and strands the extra work. This also | ||
| // ends the self-rescheduling loop, which would otherwise keep | ||
| // consuming against a closed client for the life of the process. | ||
| if (this._shuttingDown) { |
Contributor
There was a problem hiding this comment.
I think the assumption is that consume() should have no effect during drain (and shutdown, by extension), as we are already unsubscribed.
So this should not be needed, as far as we know.
- c.f. previous PR: if we need to do it for shutdown, probably also needed during "regular" rebalance.
- Should not be needed I think, but maybe there could be race conditions (if our timeout hits after kafka rebalance timeout), so it may be a good defensive measure to avoid any race condition
- On the other hand, this call is only way to
pollkafka, i.e. let it know the consumer is alive (even if not consuming). Not sure if this has any effect on rebalance, but we should make sure that skipping these calls does not actually degrade the situation by getting kafka to kick the consumer out of the group for aggressively...
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.
close()drains the in-flight work before releasing the partitions, but nothing stopped the fetch loop while it waited. Every completed task re-armed_tryConsume(), so the pipeline refilled as fast as it emptied and the departure waited on work that arrived after the shutdown began.flowchart TD C["close() begins draining"] --> W["wait for the queue to idle<br/>and the ledger to empty"] W --> T["a task completes"] T --> G{"shutting down?"} G -->|"before: not checked"| P["_tryConsume re-arms<br/>and fetches more"] P --> Q["queue refills as fast as it empties"] Q --> W G -->|"after: guard returns early"| S["fetch loop ends,<br/>queue drains to empty"] S --> L["departure proceeds"]Changes
_tryConsume()returns early once the shutdown has started. The work already in flight is still drained and its offsets still committed — only the refill stops.Measured against a 3000 message backlog:
close()The same guard ends the self-rescheduling consume loop, which otherwise kept polling a closed client for the lifetime of the process.
Verification
Two unit tests: the loop still fetches while the consumer is running, and stops once the shutdown has started. Both were checked against the previous implementation to confirm the second fails there.
End-to-end measurement of the whole stack is in #2819.
Issue: BB-833