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(),