From 653f932092448cda42b70bbcf3aa224bc3df78b8 Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Fri, 28 Aug 2026 18:04:16 +0200
Subject: [PATCH] Stop consuming once the shutdown has started

close() drains the in-flight work before releasing the partitions, but
nothing stopped the fetch loop while it waited: every completed task
re-armed _tryConsume(), so the pipeline refilled as fast as it drained
and the departure was delayed by work that arrived after the shutdown
had begun. Measured against a 3000 message backlog, close() took 6.2s
and started 301 further tasks at concurrency 10, and 9.5s and 1864
further tasks with shorter ones; with the guard both are 0 further
tasks, in 175ms and 31ms.

The same guard ends the self-rescheduling consume loop, which otherwise
kept polling a closed client for the lifetime of the process.

Issue: BB-833
---
 lib/BackbeatConsumer.js        |  8 ++++++++
 tests/unit/backbeatConsumer.js | 31 +++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js
index 915582b65..3a27f24ea 100644
--- a/lib/BackbeatConsumer.js
+++ b/lib/BackbeatConsumer.js
@@ -438,6 +438,14 @@ class BackbeatConsumer extends EventEmitter {
             this._tryConsumedTimeout = null;
         }
 
+        // the shutdown drains what is already in flight, so fetching more
+        // only delays the departure and strands the extra work. This also
+        // ends the self-rescheduling loop, which would otherwise keep
+        // consuming against a closed client for the life of the process.
+        if (this._shuttingDown) {
+            return undefined;
+        }
+
         // use non-flowing mode of consumption to add some flow
         // control: explicit consumption of messages is required,
         // needs explicit polling to get new messages
diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js
index 2ce1af459..a785337f2 100644
--- a/tests/unit/backbeatConsumer.js
+++ b/tests/unit/backbeatConsumer.js
@@ -315,6 +315,37 @@ describe('backbeatConsumer', () => {
         });
     });
 
+    describe('_tryConsume', () => {
+        let consumer;
+
+        beforeEach(() => {
+            consumer = new BackbeatConsumerMock({
+                kafka,
+                groupId: 'unittest-group',
+                topic: 'my-test-topic',
+            });
+            consumer._processingQueue = {
+                length: () => 0,
+                running: () => 0,
+            };
+            consumer._consumer = { consume: sinon.stub() };
+        });
+
+        it('should fetch while the consumer is running', () => {
+            consumer._tryConsume();
+
+            assert.strictEqual(consumer._consumer.consume.calledOnce, true);
+        });
+
+        it('should not fetch once the shutdown has started', () => {
+            consumer._shuttingDown = true;
+
+            consumer._tryConsume();
+
+            assert.strictEqual(consumer._consumer.consume.called, false);
+        });
+    });
+
     describe('_getAvailableSlotsInPipeline', () => {
         let consumer;