Stop consuming once the shutdown has started

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

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;