Answer every rebalance callback raised during shutdown

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;

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