diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js
index 566c290f3..430275fdd 100644
--- a/lib/BackbeatConsumer.js
+++ b/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());
+        });
     }
 
     /**
diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js
index 2dfa2b935..228f7948b 100644
--- a/tests/unit/backbeatConsumer.js
+++ b/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