From 44ebdf4a8267f43272703ad71806688c7f2ddd11 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Fri, 28 Aug 2026 18:04:18 +0200 Subject: [PATCH] Answer every rebalance callback raised during shutdown librdkafka requires every rebalance callback to be answered, and the shutdown path answered none of them. Each case fails differently, and each leaves the client parked in the rebalance, so the disconnect wedges on it and the process can never exit. A revoke was left for close() to answer by un-assigning later. That holds for the revoke unsubscribe() itself raises, but one arriving after close() has already un-assigned has nothing left to answer it. Answering here does not cut the drain short: close() still waits for the in-flight work, the partitions are just handed back sooner. A grant was accepted, which left the disconnect an assignment to revoke all over again. It is declined instead -- but assign() is refused once disconnect() has started, since the binding gates it on isConnected() while permitting unassign() for the whole close, so the decline falls back to unassign(). librdkafka coerces an assign into a full unassign during termination anyway. Measured across 8 full lib-suite runs per arm on CI: whenever the disconnect bound fired the process failed to exit, 40 times out of 40. Answering the callbacks takes the suite from 1 of 8 runs exiting to 8 of 8, matching 9.5 itself, with no message lost and nothing committed past unprocessed work across 240 iterations. Issue: BB-833 --- lib/BackbeatConsumer.js | 35 +++++++++++++++++++++++++++++----- tests/unit/backbeatConsumer.js | 34 ++++++++++++++++++++++++++++++--- 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index 3a27f24ea..566c290f3 100644 --- a/lib/BackbeatConsumer.js +++ b/lib/BackbeatConsumer.js @@ -819,10 +819,29 @@ class BackbeatConsumer extends EventEmitter { if (err.code === kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) { this._log.info('rdkafka.assign', { assignment }); - if (!this._shuttingDown) { - this._setDrain(null); + // close() has already handed the partitions back. Accepting a + // fresh grant now leaves the client holding an assignment the + // disconnect has to revoke all over again, which wedges it and + // costs us the LeaveGroup the shutdown exists to send. + if (this._shuttingDown) { + // assign() is refused once disconnect() has started -- the + // binding gates it on isConnected(), which is already false -- + // while unassign() is explicitly permitted while closing, and + // librdkafka coerces an assign into a full unassign during + // termination anyway. Leaving the callback unanswered is what + // parks the client in the rebalance. + this._bestEffort('assign.declined', () => { + try { + this._consumer.assign([]); + } catch (e) { // eslint-disable-line no-unused-vars + this._consumer.unassign(); + } + }); + return; } + this._setDrain(null); + try { this._consumer.assign(assignment); if (this._circuitBreaker.state !== BreakerState.Nominal) { @@ -841,10 +860,16 @@ class BackbeatConsumer extends EventEmitter { ledger: this._offsetLedger.getProcessingCount(this._topic), }); - // close() owns the departure: it drains, then un-assigns, which - // is what answers this revoke. Releasing here would cut that drain - // short and strand the offsets it exists to commit. + // librdkafka requires every rebalance callback to be answered, + // and leaving this one to close() is not enough: a revoke raised + // after close() has already un-assigned has nothing left to answer + // it, so the client stays in the rebalance and the disconnect + // wedges on it. Answering here does not cut the drain short -- + // close() still waits for the in-flight work, the partitions are + // simply given back sooner. if (this._shuttingDown) { + this._bestEffort('unassign.shutdown', + () => this._consumer.unassign()); KafkaBacklogMetrics.onRebalance( this._topic, this._groupId, unassignStatus.SHUTDOWN); return; diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js index a785337f2..2dfa2b935 100644 --- a/tests/unit/backbeatConsumer.js +++ b/tests/unit/backbeatConsumer.js @@ -586,12 +586,39 @@ describe('backbeatConsumer', () => { } }); - it('should leave a revoke arriving during shutdown to close()', () => { + it('should decline a partition grant arriving during shutdown', () => { + consumer._shuttingDown = true; + consumer._onRebalance(ASSIGN, partitions); + + // taking the grant would leave the disconnect an assignment to + // revoke all over again, which is what wedges it + assert(consumer._consumer.assign.calledOnce); + assert.deepStrictEqual( + consumer._consumer.assign.firstCall.args[0], []); + }); + + it('should still answer the grant when assign is refused mid-close', + () => { + // the binding gates assign() on isConnected(), which is already + // false once disconnect() has started, so the decline throws; + // unassign() stays permitted and is a valid answer + consumer._shuttingDown = true; + consumer._consumer.assign = sinon.stub().throws( + new Error('KafkaConsumer is not connected')); + + consumer._onRebalance(ASSIGN, partitions); + + assert(consumer._consumer.unassign.calledOnce); + }); + + it('should answer a revoke arriving during shutdown', () => { consumer._shuttingDown = true; consumer._onRebalance(REVOKE, partitions); - // releasing here would cut short the drain close() is waiting on - assert(consumer._consumer.unassign.notCalled); + // an unanswered rebalance callback leaves the client in the + // rebalance and wedges the disconnect; close() cannot be relied + // on to answer one raised after it has already un-assigned + assert(consumer._consumer.unassign.calledOnce); assert.strictEqual(consumer._drainCallback, null); assert.strictEqual(consumer._drainProcessQueueTimeout, null); assert(KafkaBacklogMetrics.onRebalance.calledWith( @@ -667,6 +694,7 @@ describe('backbeatConsumer', () => { onDisconnected = null; consumer._consumer = { + assign: sinon.stub(), commit: sinon.stub(), unassign: sinon.stub(), unsubscribe: sinon.stub(),