From 46d1b2625bee316923d30f6050853fce83ece583 Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Fri, 28 Aug 2026 18:04:20 +0200
Subject: [PATCH] Answer every caller when close() is called more than once

The services install their SIGTERM handlers with process.on rather than
once, so a repeated signal calls close() again. The second call started
its own drain wait, overwriting the single drain slot the first was
waiting on, and the first caller was then only released by its own
timeout, minutes after the consumer had already left the group.

Coalesce instead: the first call runs the shutdown, later ones attach to
it, and every caller is answered once it completes.

Issue: BB-833
---
 lib/BackbeatConsumer.js        | 13 ++++++++++++-
 tests/unit/backbeatConsumer.js | 27 +++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)

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