Leave the consumer group explicitly on shutdown - #2819
Draft
delthas wants to merge 8 commits into
Draft
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files
... and 1 file with indirect coverage changes
@@ Coverage Diff @@
## improvement/BB-835/rebalance-guard #2819 +/- ##
======================================================================
+ Coverage 75.57% 75.59% +0.01%
======================================================================
Files 200 200
Lines 13941 13991 +50
======================================================================
+ Hits 10536 10576 +40
- Misses 3395 3405 +10
Partials 10 10
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
delthas
force-pushed
the
improvement/BB-833/leave-group-on-shutdown
branch
8 times, most recently
from
August 26, 2026 15:20
f6faeab to
51e6b5d
Compare
delthas
force-pushed
the
improvement/BB-833/leave-group-on-shutdown
branch
from
August 27, 2026 10:05
7c660cd to
c2b1156
Compare
close() unsubscribed and then waited for the rebalance callback to un-assign before disconnecting. librdkafka delivers no such callback when the consumer holds no assignment, and postpones the unsubscribe outright while a rebalance is in progress, so close() never returned: the pod was SIGKILLed at the end of its grace period with the member still registered at the broker. The coordinator then kept the group waiting for a process that no longer existed, and could elect it leader of the next generation, leaving every member without partitions until the session timeout evicted it. Leave the group from close() itself, in the order the group state machine needs: commit, unsubscribe, un-assign, disconnect. Un-assigning does not clear the group assignment, so unsubscribing first is what parks the protocol on an un-assign, and our own un-assign then completes it and sends the LeaveGroup, before disconnect() and without depending on a revoke callback reaching us mid-close. In-flight work is still drained first so its offsets are committed, with the bound it already had through the revoke path, and skipped once the client is disconnected, when there is nothing left to commit and no partitions to give back. A revoke arriving during the shutdown is left for close() to answer rather than releasing the partitions early, which would cut that drain short and strand the offsets it exists to commit. Deferred un-assigns from before the shutdown are superseded for the same reason, and a partition grant arriving mid-close no longer tears down the drain close() is waiting on. Also stop waiting on an in-flight backlog publish, which could keep close() rescheduling itself indefinitely, and bound the final disconnect so a client that will not finish tearing down cannot hold the process after the group has already been left. Issue: BB-833
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
The services install their SIGTERM handlers with process.on rather than once, so a repeated signal calls close() again. The second call started its own drain wait, overwriting the single drain slot the first was waiting on, and the first caller was then only released by its own timeout, minutes after the consumer had already left the group. Coalesce instead: the first call runs the shutdown, later ones attach to it, and every caller is answered once it completes. Issue: BB-833
librdkafka delegates assignment to the rebalance callback and requires it to call assign(NULL) when the event is neither an assign nor a revoke, to resynchronise state. Ours only logged. That was harmless while the callback kept no state of its own, but it now supersedes any deferred un-assign on its way past and cancels the watchdog that would have forced one: the consumer would keep partitions it was told to release, with nothing left to notice. Un-assign instead, unless a shutdown is already under way, where close() does it. Issue: BB-833
close() hands the partitions back, but the rebalance callback still accepted a fresh grant afterwards, leaving the client holding an assignment the disconnect then had to revoke all over again. That wedged the disconnect: it hit the 5s bound, close() returned without a LeaveGroup having been sent, and the surviving members waited out session.timeout.ms instead. Measured across 288 CI iterations against a real broker, this accounted for every remaining degraded departure on the fixed tree -- 6 of 69, all in the rejoining and draining states, median takeover 29.9s against 3.0s when it did not fire, three of them at 40-44s which is eviction rather than a departure. Issue: BB-833
librdkafka requires every rebalance callback to be answered, and leaving this one for close() to answer by un-assigning later is not enough: a revoke raised after close() has already un-assigned has nothing left to answer it. The client then stays in the rebalance, and the disconnect waits on it until its bound gives up -- which returns from close() looking successful while leaving a client whose destructor blocks, so the process can never exit. Measured across 8 full lib-suite runs per arm on CI, the correlation between that bound firing and the process failing to exit was 40 out of 40. Answering the callback here takes the suite from 1 of 8 runs exiting to 8 of 8, matching 9.5 itself. It does not cut the drain short, which is what the previous comment feared: close() still waits for the in-flight work, the partitions are just handed back sooner. Across 240 iterations the draining scenario keeps the same ~5.95s close and the same volume processed, with no message lost, nothing committed past unprocessed work, and fewer duplicates than before. Answering only once close() had released was also measured, and fixes almost nothing (2 of 8): the revokes that matter arrive before that. Issue: BB-833
delthas
force-pushed
the
improvement/BB-833/leave-group-on-shutdown
branch
from
August 27, 2026 16:44
fe56c8d to
829927b
Compare
Declining a partition grant during shutdown with assign([]) only works until disconnect() starts. The binding gates assign() on isConnected(), which Disconnect() clears before it takes its lock, while unassign() is gated on !IsClosing() && !IsConnected() and so stays permitted for the whole close. Since the fetch loop stops as soon as the shutdown begins, the only thing still delivering rebalance callbacks by then is the closing client itself -- so the decline was refused exactly when it was needed, and swallowed by the best-effort wrapper. It does not hang today: librdkafka's terminate0 unsubscribes on its own and rejoins to INIT, which recovers the state. But that is the library covering for us, and leaving a rebalance callback unanswered is the same defect class as the revoke path fixed one commit earlier. librdkafka coerces an assign into a full unassign once termination has started, so unassign() is a valid answer to a grant here. Issue: BB-833
The replication queue processor's stop() closes its status producer before the consumers, and a task's offset only becomes committable from that producer's delivery callback -- so a report that never arrives leaves a ledger entry outstanding while the processing queue is idle. Measured against a permanently stuck ledger, the previous close() never returned at all; this one returns in maxPollIntervalMs - 1000 plus the disconnect bound. The existing drain test holds both signals, so this covers the ledger-only shape the real service produces. Issue: BB-833
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
1 participant
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()unsubscribed and then waited for the rebalance callback to un-assign before disconnecting. librdkafka delivers no such callback when the consumer holds no assignment, and postpones the unsubscribe outright while a rebalance is in progress — soclose()never returned and the pod was SIGKILLed with the member still registered at the broker.sequenceDiagram participant C as BackbeatConsumer participant K as Kafka Note over C: SIGTERM during a rebalance C->>K: unsubscribe() Note right of K: postponed — a rebalance<br/>is already in progress C->>C: wait for 'unassign' … forever Note over C: SIGKILL at the grace period,<br/>no LeaveGroup ever sent Note over K: member still registered,<br/>may be elected leader of the next<br/>generation and never SyncGroupThe group then holds zero partitions until
session.timeout.ms(45 s) evicts the member. During a rolling update a rebalance is in progress essentially by construction, since the new pod joins before the old one is told to stop.Changes
Release the partitions and drop the subscription before closing, so the close path has nothing to hand back to us and the
LeaveGroupgoes out whatever state the group is in:The bracketed steps only ran if librdkafka delivered a revoke callback. It delivers none when the consumer holds no assignment, and postpones the unsubscribe outright while a rebalance is in progress — in both cases the wait never ends. The same steps now run unconditionally, in
close()itself.The order of the last two matters, and the mechanism is a flag rather than the assignment list. Only
rd_kafka_cgrp_unsubscribe()setsF_LEAVE_ON_UNASSIGN_DONE, and onlyunassign_done()— reached fromWAIT_UNASSIGN_CALL— consults it to send theLeaveGroup. Un-assigning first therefore clears the assignment without any state change and theLeaveGroupis never armed;unsubscribe()then fires a revoke at us and parks inWAIT_UNASSIGN_CALL, leaving theLeaveGroupgated on a callback round trip thatdisconnect()is simultaneously blocking on. Unsubscribing first puts us in the one join-state whereunassign()is meaningful, so our own call completes it and sends theLeaveGroupbeforedisconnect()is reached.Draining also means we must stop fetching. Nothing did: 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. 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.
close()to answer by un-assigning later is not enough: one raised afterclose()has already un-assigned has nothing left to answer it, so the client stays in the rebalance and the disconnect wedges on it. Answering does not cut the drain short —close()still waits for the in-flight work, the partitions are simply handed back sooner, and the offsets that drain exists to commit (BB-758) are unaffected.close()installed. Both previously leftclose()waiting on a callback that could never fire.close()no longer waits on an in-flight backlog publish, which could otherwise keep it rescheduling itself every second indefinitely.close()has released the partitions is declined rather than accepted, so the disconnect is not handed an assignment it must revoke all over again.close()looking successful while leaving a client whose destructor blocks. Answering the rebalance callback above is what stops it firing; the bound is no longer load-bearing.In-flight work is still drained before the partitions are released, so offsets are committed exactly as before, and that wait keeps the bound it already had through the revoke path (
max.poll.interval.ms - 1000) — a wedged task delays the departure no longer than it does today. Draining is skipped once the client is disconnected, since there is then nothing to commit and no partitions to give back.That bound is inherited, not chosen: at the default
max.poll.interval.msit is ~299 s, far longer than a pod's grace period, so a wedged task is still killed rather than departing cleanly. Replacing it with a deadline derived from the grace period is the budget work, deliberately left out here — BB-854 shortens the drain first.Verification
Unit tests for the call ordering, completion with no assignment held, the drain wait, the offset-publish skip, watchdog cleanup, and a revoke arriving mid-drain. Each was checked against the previous implementation to confirm it fails there.
Two functional tests against a real broker. The second reproduces the incident: a newcomer joins, and the member being closed has already released its partitions and is waiting to rejoin, so the rebalance is still in progress and nothing will revoke back to it.
close()never returnsThe first passes either way — it guards the
unsubscribe→unassignordering, since reverting that gates theLeaveGroupon a callbackdisconnect()is blocking on and the takeover falls off the 45 s cliff. The second is the regression guard.Whether the process then exits was measured separately, 8 full runs of the lib suite per tree: 9.5 exits 8/8, this branch 8/8. Before the callback fix it was 1/8, and every wedged run had the disconnect bound firing first.
Beyond that, a pod-level census on real CI runners: a pod is terminated in each of four states, and what both pods actually processed is reconciled against what was produced. 288 iterations per round, two arms measured by identical harness code with only
lib/differing.n = 62 / 77 valid samples. Across the two rounds (290 valid samples) no iteration lost a message, and none ever committed an offset past work it had not finished — the property that matters more than the timing.
The census also found a defect that review had not: the rebalance callback still accepted a partition grant after
close()had handed the partitions back, so the disconnect had to revoke them again, wedged, hit its 5 s bound, and returned without aLeaveGroup— the surviving members then waited outsession.timeout.ms. That was 6 of 69 iterations; declining the grant took the SIGKILL rate to zero and halved the residual.What remains is ~4% of departures still landing at 40-45 s, i.e. eviction rather than a departure. That signature is the orphaned member id tracked upstream as BB-843, not this path, but that attribution is a hypothesis rather than something these runs establish.
Issue: BB-833