Leave the consumer group explicitly on shutdown

[delthas]

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

Loading

The 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

Extract the drain predicate into a method — no behaviour change, same predicate, same two call sites. The revoke path already tested that the processing queue and the offset ledger are both drained; the shutdown path needs the same test, so it is lifted out of the revoke closure rather than restated.

Hoist the functional waitFor helper to module scope — also no behaviour change. It lived inside the fromOffset suite; the departure tests below need the same poll-until-true wait, so it moves out and hands back a promise rather than taking a callback.

Leave the consumer group explicitly on shutdown. Drain, commit, unsubscribe and un-assign now all run unconditionally in close() itself, rather than in a revoke callback that may never arrive. The close path is then handed nothing it has to give back:

sequenceDiagram
    participant C as BackbeatConsumer
    participant L as librdkafka cgrp
    participant K as Kafka
    C->>C: drain in-flight work
    C->>K: commit offsets
    C->>L: unsubscribe()
    Note right of L: arms F_LEAVE_ON_UNASSIGN_DONE,<br/>parks in WAIT_UNASSIGN_CALL
    C->>L: unassign()
    Note right of L: unassign_done() reaches the flag
    L->>K: LeaveGroup
    C->>L: disconnect()
    Note over C,K: group already left,<br/>nothing left to revoke

Loading

The order of the last two matters, and the mechanism is a flag rather than the assignment list: only rd_kafka_cgrp_unsubscribe() sets F_LEAVE_ON_UNASSIGN_DONE, and only unassign_done() consults it. Un-assigning first clears the assignment with no state change, so the flag is never armed — unsubscribe() then fires a revoke at us and parks, leaving the LeaveGroup gated on a callback round trip that disconnect() is simultaneously blocking on.

In-flight work is still drained before the partitions are released, so offsets are committed exactly as before (BB-758), 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.ms it 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

Fourteen unit tests covering the new call ordering, completion when no consumer was ever created, the drain wait and its bound, the offset-publish skip, a revoke or a grant arriving mid-close, and a step throwing part-way through. 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.

	before	after
close with the group settled	3994 ms	3994 ms
close during a rebalance	close() never returns	10008 ms

The first passes either way — it guards the unsubscribe → unassign ordering, since reverting that gates the LeaveGroup on a callback disconnect() is blocking on and the takeover falls off the 45 s cliff. The second is the regression guard.

Measured across the whole series

These numbers come from the whole stack — this PR plus #2834, #2835 and #2836 — not from this PR alone. 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.

	whole stack	before
exits without being SIGKILLed	100%	0%
median takeover	3.3 s	16.3 s
takeover over 10 s	9%	100%
takeover over 30 s	4%	12%
messages lost	0	0

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.

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

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 75.57%. Comparing base (6c6acda) to head (22f3a61).

Additional details and impacted files

Files with missing lines	Coverage Δ
lib/BackbeatConsumer.js	94.61% <100.00%> (+0.80%)	⬆️
lib/constants.js	100.00% <ø> (ø)

Components	Coverage Δ
Bucket Notification	80.27% <ø> (ø)
Core Library	81.42% <100.00%> (+0.12%)	⬆️
Ingestion	70.09% <ø> (ø)
Lifecycle	80.46% <ø> (ø)
Oplog Populator	85.83% <ø> (ø)
Replication	62.01% <ø> (ø)
Bucket Scanner	85.76% <ø> (ø)

@@                          Coverage Diff                           @@
##           improvement/BB-835/rebalance-guard    #2819      +/-   ##
======================================================================
+ Coverage                               75.51%   75.57%   +0.06%     
======================================================================
  Files                                     200      200              
  Lines                                   13946    13974      +28     
======================================================================
+ Hits                                    10531    10561      +30     
+ Misses                                   3405     3403       -2     
  Partials                                   10       10              

Flag	Coverage Δ
api:retry	9.07% <0.00%> (-0.02%)	⬇️
api:routes	8.84% <0.00%> (-0.02%)	⬇️
bucket-scanner	85.76% <ø> (ø)
ft_test:queuepopulator	9.17% <4.16%> (-0.01%)	⬇️
ingestion	12.22% <2.08%> (-0.02%)	⬇️
lifecycle	19.41% <62.50%> (+0.08%)	⬆️
notification	1.01% <0.00%> (-0.01%)	⬇️
oplogPopulator	0.13% <0.00%> (-0.01%)	⬇️
replication	18.96% <62.50%> (+0.08%)	⬆️
unit	55.49% <100.00%> (+0.25%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[delthas]

Requested @maeldonn in place of Sylvain Senechal, who is currently on PTO.

[francoisferrand]

From what I understand, the issue would be if we get a shutdown request during rebalance, where the following guard is not (always) working properly:

                if (this._consumer?.isConnected()) {
                    const subscription = this._getSubscription();
                    if (subscription !== null) {
                        this._consumer.unsubscribe();
                        // Wait for partition unassign to complete before
                        // disconnecting, the rebalance callback will handle
                        // waiting for current jobs to complete as well as commit
                        // the latest offsets
                        this.once('unassign', () => next());

i.e. we don't have the guarantee (from librdkafka) that calling unsubstribe() will always trigger the rebalance revoke event...

But since we send unassign ourself, this would actually be easy to work around, by having some kind of state machine (in our code) : i.e. if we are rebalancing, we just remember we are also closing so can call next when done, etc...

maybe just something like:

                if (this._consumer?.isConnected()) {
                    const subscription = this._getSubscription();
                    if (subscription !== null) {
                        this._consumer.unsubscribe();
                    }
                    if (subscription !== null || rebalance_in_progress) {
                        // Wait for partition unassign to complete before
                        // disconnecting, the rebalance callback will handle
                        // waiting for current jobs to complete as well as commit
                        // the latest offsets
                        this.once('unassign', () => next());

(the whole change in this PR seems awfully complex and redundant, having 2 separate "drain" path... so I wonder if we are really fixing the issue at the root, or just adding lot of code to try and cope...)