Answer every caller when close() is called more than once

lib/BackbeatConsumer.js

@@ -181,6 +181,7 @@ class BackbeatConsumer extends EventEmitter {
         // a deferred un-assign gives up when this no longer matches
         this._rebalanceId = 0;
         this._shuttingDown = false;
+        this._closeCallbacks = null;
 
         this._messagesConsumed = 0;
         // this variable represents how many kafka messages have been
@@ -1341,6 +1342,12 @@ class BackbeatConsumer extends EventEmitter {
      * @return {undefined}
      */
     close(cb) {
+        if (this._closeCallbacks) {
+            this._closeCallbacks.push(cb);
+            return undefined;
+        }
+        this._closeCallbacks = [cb];
+
         if (this._publishOffsetsCronTimer) {
             clearInterval(this._publishOffsetsCronTimer);
             this._publishOffsetsCronTimer = null;
@@ -1393,7 +1400,11 @@ class BackbeatConsumer extends EventEmitter {
                 this._consumer.disconnect();
                 return undefined;
             },
-        ], () => cb());
+        ], () => {
+            const callbacks = this._closeCallbacks;
+            this._closeCallbacks = null;
+            callbacks.forEach(done => done());
+        });
     }
 
     /**

tests/unit/backbeatConsumer.js

@@ -881,6 +881,33 @@ describe('backbeatConsumer', () => {
             });
         });
 
+        it('should answer every caller once when closed twice', done => {
+            queueIdle = false;
+            ledgerCount = 1;
+
+            let first = false;
+            let second = 0;
+            consumer.close(() => {
+                first = true;
+            });
+            consumer.close(() => {
+                second++;
+            });
+
+            setImmediate(() => {
+                queueIdle = true;
+                ledgerCount = 0;
+                drainCallbacks[drainCallbacks.length - 1]();
+
+                setTimeout(() => {
+                    assert.strictEqual(first, true);
+                    assert.strictEqual(second, 1);
+                    assert(consumer._consumer.disconnect.calledOnce);
+                    done();
+                }, 20);
+            });
+        });
+
         it('should not wait for the drain once the consumer is disconnected', done => {
             // the drain watchdog fires on a wedged task and disconnects, so the
             // work it is waiting on can never complete