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;