diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index e0fd50d8a..915582b65 100644 --- a/lib/BackbeatConsumer.js +++ b/lib/BackbeatConsumer.js @@ -28,6 +28,8 @@ const tracing = require('./tracing'); const { startLinkedSpanFromKafkaEntry } = require('arsenal/build/lib/tracing').kafka; const CLIENT_ID = 'BackbeatConsumer'; +// the group has already been left by the time we disconnect +const DISCONNECT_TIMEOUT_MS = 5000; const { withTopicPrefix } = require('./util/topic'); /** @@ -178,6 +180,7 @@ class BackbeatConsumer extends EventEmitter { // a deferred un-assign gives up when this no longer matches this._rebalanceId = 0; + this._shuttingDown = false; this._messagesConsumed = 0; // this variable represents how many kafka messages have been @@ -755,6 +758,22 @@ class BackbeatConsumer extends EventEmitter { } } + /** + * Both the processing queue and the offset ledger must be drained before + * releasing partitions. The queue covers in-flight worker invocations + * (committable: true path); the ledger covers entries still waiting for + * their deferred onEntryCommittable to land (committable: false path — + * typically a Kafka status producer's delivery callback). Releasing the + * partition before the ledger drains would cause offsetsStore to hit + * ERR__STATE on the deferred callbacks (BB-758). + * + * @returns {boolean} true if there is no work left in flight + */ + _isFullyDrained() { + return this._processingQueue.idle() && + this._offsetLedger.getProcessingCount(this._topic) === 0; + } + /** * Run a shutdown/rebalance step that must not abort the sequence it * belongs to, logging rather than throwing. @@ -792,7 +811,9 @@ class BackbeatConsumer extends EventEmitter { if (err.code === kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) { this._log.info('rdkafka.assign', { assignment }); - this._setDrain(null); + if (!this._shuttingDown) { + this._setDrain(null); + } try { this._consumer.assign(assignment); @@ -812,17 +833,32 @@ class BackbeatConsumer extends EventEmitter { ledger: this._offsetLedger.getProcessingCount(this._topic), }); - const isSuperseded = () => rebalanceId !== this._rebalanceId; + // close() owns the departure: it drains, then un-assigns, which + // is what answers this revoke. Releasing here would cut that drain + // short and strand the offsets it exists to commit. + if (this._shuttingDown) { + KafkaBacklogMetrics.onRebalance( + this._topic, this._groupId, unassignStatus.SHUTDOWN); + return; + } + + const isSuperseded = () => + rebalanceId !== this._rebalanceId || this._shuttingDown; const skipSuperseded = status => { - this._log.info('skipping superseded un-assign', { + // close() answers the revoke itself once it has drained, so + // a shutdown is a distinct reason from a later rebalance + const reason = this._shuttingDown ? + unassignStatus.SHUTDOWN : unassignStatus.SUPERSEDED; + this._log.info('skipping deferred un-assign', { status, + reason, rebalanceId, currentRebalanceId: this._rebalanceId, topic: this._topic, groupId: this._groupId, }); KafkaBacklogMetrics.onRebalance( - this._topic, this._groupId, unassignStatus.SUPERSEDED); + this._topic, this._groupId, reason); }; const unassign = jsutil.once(status => { @@ -883,18 +919,7 @@ class BackbeatConsumer extends EventEmitter { } }); - // Both the processing queue and the offset ledger must be drained - // before unassigning. The queue covers in-flight worker invocations - // (committable: true path); the ledger covers entries still waiting - // for their deferred onEntryCommittable to land (committable: false - // path — typically a Kafka status producer's delivery callback). - // Releasing the partition before the ledger drains would cause - // offsetsStore to hit ERR__STATE on the deferred callbacks (BB-758). - const isFullyDrained = () => - this._processingQueue.idle() && - this._offsetLedger.getProcessingCount(this._topic) === 0; - - if (isFullyDrained()) { + if (this._isFullyDrained()) { unassign(unassignStatus.IDLE); return; } @@ -905,7 +930,7 @@ class BackbeatConsumer extends EventEmitter { // checkFullyDrained re-checks both conditions and only // triggers unassign once both hold. this._setDrain(() => { - if (isFullyDrained()) { + if (this._isFullyDrained()) { unassign(unassignStatus.DRAINED); } }); @@ -1287,39 +1312,100 @@ class BackbeatConsumer extends EventEmitter { clearInterval(this._publishOffsetsCronTimer); this._publishOffsetsCronTimer = null; } - if (this._publishOffsetsCronActive) { - return setTimeout(() => this.close(cb), 1000); - } + // its watchdog would otherwise disconnect us mid-departure + clearTimeout(this._drainProcessQueueTimeout); + this._drainProcessQueueTimeout = null; + this._shuttingDown = true; this._circuitBreaker.stop(); + return async.waterfall([ next => { - if (this._consumer?.isConnected()) { - const subscription = this._getSubscription(); - if (subscription !== null) { - this._consumer.unsubscribe(); - // Wait for partition unassign to complete before - // disconnecting, the rebalance callback will handle - // waiting for current jobs to complete as well as commit - // the latest offsets - this.once('unassign', () => next()); - return; - } + // draining buys a commit before the partitions go, which is + // pointless once the client is gone + if (!this._consumer?.isConnected()) { + return process.nextTick(next); } - process.nextTick(next); + return this._drainBeforeShutdown(next); }, next => { if (this._zookeeper) { this._zookeeper.close(); } - if (this._consumer?.isConnected()) { - this._consumer.disconnect(); - this._consumer.once('disconnected', () => next()); - } else { - process.nextTick(next); + if (!this._consumer?.isConnected()) { + return process.nextTick(next); } + // commit first: un-assigning resets the stored offsets. + // unsubscribe next, which leaves the group protocol waiting on + // an un-assign, and that un-assign is what sends the + // LeaveGroup — before disconnect(), and without depending on + // a revoke callback reaching us mid-close. + this._bestEffort('commit', () => this._consumer.commit()); + this._bestEffort('unsubscribe', () => this._consumer.unsubscribe()); + this._bestEffort('unassign', () => this._consumer.unassign()); + + const disconnected = jsutil.once(next); + const timer = setTimeout(() => { + this._log.warn('consumer did not finish disconnecting, ' + + 'exiting anyway', { + timeoutMs: DISCONNECT_TIMEOUT_MS, + topic: this._topic, + groupId: this._groupId, + }); + disconnected(); + }, DISCONNECT_TIMEOUT_MS); + this._consumer.once('disconnected', () => { + clearTimeout(timer); + disconnected(); + }); + this._consumer.disconnect(); + return undefined; }, ], () => cb()); } + + /** + * Wait for in-flight work so its offsets are committed before the + * partitions go, bounded like the revoke path so a wedged task cannot + * hold the departure any longer than it already did. + * + * @param {function} cb - callback + * @returns {undefined} + */ + _drainBeforeShutdown(cb) { + const done = jsutil.once(cb); + if (this._isFullyDrained()) { + return process.nextTick(done); + } + this._log.info('waiting for in-flight work before leaving the group', { + queueLen: this._processingQueue.length(), + running: this._processingQueue.running(), + ledger: this._offsetLedger.getProcessingCount(this._topic), + topic: this._topic, + groupId: this._groupId, + }); + // same bound the revoke path uses, so a wedged task delays the + // departure no longer than it did before + const timer = setTimeout(() => { + this._log.warn('giving up on in-flight work, leaving the group', { + queueLen: this._processingQueue.length(), + running: this._processingQueue.running(), + ledger: this._offsetLedger.getProcessingCount(this._topic), + topic: this._topic, + groupId: this._groupId, + }); + this._setDrain(null); + done(); + }, this._maxPollIntervalMs - 1000); + + this._setDrain(() => { + if (this._isFullyDrained()) { + clearTimeout(timer); + this._setDrain(null); + done(); + } + }); + return undefined; + } } module.exports = BackbeatConsumer; diff --git a/lib/constants.js b/lib/constants.js index 183359b29..079d765a9 100644 --- a/lib/constants.js +++ b/lib/constants.js @@ -26,6 +26,7 @@ const constants = { DRAINED: 'drained', TIMEOUT: 'timeout', SUPERSEDED: 'superseded', + SHUTDOWN: 'shutdown', }, statusReady: 'READY', statusUndefined: 'UNDEFINED', diff --git a/tests/functional/lib/BackbeatConsumer.js b/tests/functional/lib/BackbeatConsumer.js index 060cb3aee..a92aa2f90 100644 --- a/tests/functional/lib/BackbeatConsumer.js +++ b/tests/functional/lib/BackbeatConsumer.js @@ -2,10 +2,13 @@ const assert = require('assert'); const async = require('async'); const sinon = require('sinon'); const werelogs = require('werelogs'); +const { promisify } = require('util'); +const Kafka = require('node-rdkafka'); const { metrics } = require('arsenal'); const ZookeeperManager = require('../../../lib/clients/ZookeeperManager'); +const { withTopicPrefix } = require('../../../lib/util/topic'); const BackbeatProducer = require('../../../lib/BackbeatProducer'); const BackbeatConsumer = require('../../../lib/BackbeatConsumer'); const { BreakerState, CircuitBreaker } = require('breakbeat').CircuitBreaker; @@ -24,6 +27,22 @@ const consumerKafkaConf = { }; const log = new werelogs.Logger('BackbeatConsumer:test'); +function waitFor(predicate, timeoutMs, description) { + const deadline = Date.now() + timeoutMs; + return new Promise((resolve, reject) => { + const check = () => { + if (predicate()) { + return resolve(); + } + if (Date.now() > deadline) { + return reject(new Error(`timed out waiting for ${description}`)); + } + return setTimeout(check, 200); + }; + check(); + }); +} + describe('BackbeatConsumer main tests', () => { const topic = 'backbeat-consumer-spec'; const groupId = `replication-group-${Math.random()}`; @@ -1275,6 +1294,188 @@ describe('BackbeatConsumer shutdown tests', () => { }).timeout(60000); }); +describe('BackbeatConsumer group departure tests', () => { + const topic = 'backbeat-consumer-spec-departure'; + const groupId = `bucket-processor-${Math.random()}`; + // enough partitions for every member to hold a share, so the group is + // observably settled before a departure is timed + const wantedPartitions = 3; + // a member that never sends LeaveGroup is only evicted once + // session.timeout.ms (45s) expires; leaving cleanly takes a few seconds + const takeoverBudgetMs = 30000; + const adminTimeoutMs = 20000; + const settleTimeoutMs = 60000; + const messages = Array.from({ length: 12 }, (_, i) => + ({ key: `key-${i}`, message: `{"value":"${i}"}` })); + // read back rather than assumed: a topic left over from an earlier run + // may be wider than we asked for + let partitionCount; + let admin; + let producer; + let leaving; + let survivor; + let newcomer; + // the survivor holds its tasks so its un-assign stays deferred, which is + // what keeps a rebalance in progress for as long as a test needs + let inFlight; + let holdTasks; + + function createConsumer(clientId, queueProcessor) { + return new BackbeatConsumer({ + clientId, + zookeeper: zookeeperConf, + kafka: consumerKafkaConf, + groupId, + topic, + concurrency: 1, + queueProcessor, + }); + } + + function partitionsHeld(consumer) { + // a closed client throws rather than reporting an empty assignment + return consumer._consumer.isConnected() ? + consumer._consumer.assignments().length : 0; + } + + function totalPartitionsHeld(consumers) { + return consumers.reduce( + (total, consumer) => total + partitionsHeld(consumer), 0); + } + + function completeInFlight() { + holdTasks = false; + const pending = inFlight; + inFlight = []; + pending.forEach(cb => cb()); + } + + function close(consumer) { + return new Promise(resolve => consumer.close(resolve)); + } + + before(async function before() { + this.timeout(60000); + admin = Kafka.AdminClient.create({ + 'client.id': 'kafka-admin', + 'metadata.broker.list': consumerKafkaConf.hosts, + }); + const { ERR_TOPIC_ALREADY_EXISTS, ERR_INVALID_PARTITIONS } = + Kafka.CODES.ERRORS; + try { + await promisify(admin.createTopic).bind(admin)({ + topic: withTopicPrefix(topic), + num_partitions: wantedPartitions, // eslint-disable-line camelcase + replication_factor: 1, // eslint-disable-line camelcase + }, adminTimeoutMs); + } catch (err) { + if (err.code !== ERR_TOPIC_ALREADY_EXISTS) { + throw err; + } + // left over from an earlier run, possibly narrower than we need: + // widen it rather than waiting on partitions that never come + try { + await promisify(admin.createPartitions).bind(admin)( + withTopicPrefix(topic), wantedPartitions, adminTimeoutMs); + } catch (widenErr) { + if (widenErr.code !== ERR_INVALID_PARTITIONS) { + throw widenErr; + } + } + } + }); + + after(() => admin.disconnect()); + + beforeEach(async function beforeEach() { + this.timeout(120000); + inFlight = []; + holdTasks = true; + producer = new BackbeatProducer({ + kafka: producerKafkaConf, + topic, + pollIntervalMs: 100, + }); + leaving = createConsumer('BackbeatConsumer-leaving', + (message, cb) => cb()); + survivor = createConsumer('BackbeatConsumer-survivor', + (message, cb) => (holdTasks ? inFlight.push(cb) : cb())); + newcomer = createConsumer('BackbeatConsumer-newcomer', + (message, cb) => cb()); + await Promise.all([ + new Promise(resolve => producer.on('ready', resolve)), + ...[leaving, survivor, newcomer].map(consumer => + new Promise(resolve => consumer.on('ready', resolve))), + ]); + const metadata = await promisify(leaving._consumer.getMetadata) + .bind(leaving._consumer)({ topic: leaving._topic, timeout: adminTimeoutMs }); + partitionCount = metadata.topics + .find(entry => entry.name === leaving._topic).partitions.length; + assert(partitionCount >= wantedPartitions, + `topic has ${partitionCount} partitions, need ${wantedPartitions}`); + + leaving.subscribe(); + survivor.subscribe(); + await waitFor( + () => totalPartitionsHeld([leaving, survivor]) === partitionCount, + settleTimeoutMs, 'the group to settle'); + }); + + afterEach(async function afterEach() { + this.timeout(30000); + // start closing first so the consumers stop fetching, then release + // the held tasks so the drain they are waiting on can finish + const closed = Promise.all([leaving, survivor, newcomer].map(close)); + completeInFlight(); + await closed; + await new Promise(resolve => producer.close(resolve)); + }); + + it('should leave the group so the remaining member takes over promptly', + async function departure() { + this.timeout(90000); + const start = Date.now(); + await close(leaving); + await waitFor(() => partitionsHeld(survivor) === partitionCount, + takeoverBudgetMs, 'the survivor to take over every partition'); + const elapsed = Date.now() - start; + assert(elapsed < takeoverBudgetMs, + `takeover took ${elapsed}ms: the group was not left cleanly`); + }); + + it('should leave the group when closed during a rebalance', + async function departureWhileRebalancing() { + this.timeout(90000); + // a task in flight on the survivor defers its un-assign, which holds + // the rebalance triggered below open for as long as we need + producer.send(messages, assert.ifError); + await waitFor(() => inFlight.length > 0, settleTimeoutMs, + 'the survivor to start a task'); + newcomer.subscribe(); + // the join revokes both members: `leaving` releases its partitions and + // waits to rejoin while the survivor drains, so it is closed holding + // no assignment with the rebalance still in progress -- nothing will + // revoke back to it to complete the departure + await waitFor(() => partitionsHeld(leaving) === 0 + && partitionsHeld(survivor) > 0, + settleTimeoutMs, 'the member being closed to release its partitions'); + const start = Date.now(); + await close(leaving); + const closedIn = Date.now() - start; + assert(closedIn < takeoverBudgetMs, + `close() took ${closedIn}ms while rebalancing`); + completeInFlight(); + // the newcomer must be given a share, else the survivor merely still + // holds what it had and nothing was actually handed over + await waitFor(() => partitionsHeld(newcomer) > 0 + && totalPartitionsHeld([survivor, newcomer]) === partitionCount, + takeoverBudgetMs, 'the remaining members to share every partition'); + const elapsed = Date.now() - start; + assert(elapsed < takeoverBudgetMs, + `takeover took ${elapsed}ms: the group was not left cleanly`); + }); +}); + describe('BackbeatConsumer fromOffset tests', () => { const topic = 'backbeat-consumer-spec-from-offset'; let producer; @@ -1286,20 +1487,6 @@ describe('BackbeatConsumer fromOffset tests', () => { process.nextTick(cb); } - function waitFor(predicate, timeoutMs, description, cb) { - const deadline = Date.now() + timeoutMs; - const check = () => { - if (predicate()) { - return cb(); - } - if (Date.now() > deadline) { - return cb(new Error(`timed out waiting for ${description}`)); - } - return setTimeout(check, 200); - }; - check(); - } - function startConsumer(cb) { consumer = new BackbeatConsumer({ clientId: 'BackbeatConsumer-fromOffset', @@ -1369,7 +1556,8 @@ describe('BackbeatConsumer fromOffset tests', () => { // session timeout (45s) to settle when a rebalance hits // its joining window next => waitFor(() => consumedMessages.includes(marker), 75000, - 'the pre-existing message to be consumed', next), + 'the pre-existing message to be consumed') + .then(() => next(), next), ], done); }); }); diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js index afc7d7238..2ce1af459 100644 --- a/tests/unit/backbeatConsumer.js +++ b/tests/unit/backbeatConsumer.js @@ -555,6 +555,33 @@ describe('backbeatConsumer', () => { } }); + it('should leave a revoke arriving during shutdown to close()', () => { + consumer._shuttingDown = true; + consumer._onRebalance(REVOKE, partitions); + + // releasing here would cut short the drain close() is waiting on + assert(consumer._consumer.unassign.notCalled); + assert.strictEqual(consumer._drainCallback, null); + assert.strictEqual(consumer._drainProcessQueueTimeout, null); + assert(KafkaBacklogMetrics.onRebalance.calledWith( + 'my-test-topic', 'unittest-group', unassignStatus.SHUTDOWN)); + }); + + it('should ignore a deferred un-assign once the shutdown owns the ' + + 'departure', () => { + consumer._onRebalance(REVOKE, partitions); + const deferredUnassign = drainCallbacks[drainCallbacks.length - 1]; + + // close() takes over while the drain is outstanding + consumer._shuttingDown = true; + + queueIdle = true; + ledgerCount = 0; + deferredUnassign(); + + assert(consumer._consumer.unassign.notCalled); + }); + it('should leave the current drain and timeout armed when a superseded ' + 'un-assign fires', () => { consumer._onRebalance(REVOKE, partitions); @@ -581,4 +608,271 @@ describe('backbeatConsumer', () => { assert(consumer._consumer.unassign.notCalled); }); }); + + describe('close', () => { + let consumer; + let queueIdle; + let ledgerCount; + let drainCallbacks; + let onDisconnected; + + beforeEach(() => { + consumer = new BackbeatConsumerMock({ + kafka, + groupId: 'unittest-group', + topic: 'my-test-topic', + }); + + queueIdle = true; + ledgerCount = 0; + drainCallbacks = []; + consumer._processingQueue = { + length: () => 0, + running: () => (queueIdle ? 0 : 1), + idle: () => queueIdle, + setDrain: func => drainCallbacks.push(func), + }; + consumer._offsetLedger.getProcessingCount = () => ledgerCount; + + onDisconnected = null; + consumer._consumer = { + commit: sinon.stub(), + unassign: sinon.stub(), + unsubscribe: sinon.stub(), + // only a real disconnect emits 'disconnected' + disconnect: sinon.stub().callsFake(() => { + if (onDisconnected) { + setImmediate(onDisconnected); + } + }), + assignments: () => [], + subscription: () => ['my-test-topic'], + isConnected: () => true, + once: (event, handler) => { + if (event === 'disconnected') { + setImmediate(handler); + } + }, + }; + }); + + afterEach(() => { + sinon.restore(); + }); + + it('should leave the group before disconnecting', done => { + consumer.close(() => { + const { commit, unassign, unsubscribe, disconnect } = consumer._consumer; + assert(commit.calledOnce); + assert(unassign.calledOnce); + assert(unsubscribe.calledOnce); + assert(disconnect.calledOnce); + // committing after un-assign would commit nothing, and it is + // the un-assign following unsubscribe that sends the LeaveGroup + assert(commit.calledBefore(unsubscribe)); + assert(unsubscribe.calledBefore(unassign)); + assert(unassign.calledBefore(disconnect)); + done(); + }); + }); + + it('should keep waiting for its drain when a grant arrives mid-close', done => { + queueIdle = false; + ledgerCount = 1; + + let closed = false; + consumer.close(() => { + closed = true; + }); + + setImmediate(() => { + // a grant must not tear down the wait close() installed + consumer._onRebalance( + { code: CODES.ERRORS.ERR__ASSIGN_PARTITIONS }, + [{ topic: 'my-test-topic', partition: 0 }]); + assert.strictEqual(closed, false); + + queueIdle = true; + ledgerCount = 0; + drainCallbacks[drainCallbacks.length - 1](); + + setImmediate(() => { + assert.strictEqual(closed, true); + done(); + }); + }); + }); + + it('should complete when the consumer was never created', done => { + // close() can race startup, before _initConsumer() has run + consumer._consumer = null; + consumer.close(done); + }); + + it('should keep leaving the group when a call throws', done => { + consumer._consumer.commit.throws(new Error('Local: Erroneous state')); + consumer.close(() => { + assert(consumer._consumer.unsubscribe.calledOnce); + assert(consumer._consumer.unassign.calledOnce); + assert(consumer._consumer.disconnect.calledOnce); + done(); + }); + }); + + it('should wait for in-flight work before releasing the partitions', done => { + queueIdle = false; + ledgerCount = 1; + + let closed = false; + consumer.close(() => { + closed = true; + }); + + setImmediate(() => { + assert.strictEqual(closed, false); + assert(consumer._consumer.unassign.notCalled); + + queueIdle = true; + ledgerCount = 0; + drainCallbacks[drainCallbacks.length - 1](); + + setImmediate(() => { + assert.strictEqual(closed, true); + assert(consumer._consumer.unassign.calledOnce); + done(); + }); + }); + }); + + it('should still complete when a revoke arrives while it waits for ' + + 'the drain', done => { + queueIdle = false; + ledgerCount = 1; + + let closed = false; + consumer.close(() => { + closed = true; + }); + + setImmediate(() => { + // releasing partitions here must not tear down close()'s wait + consumer._onRebalance( + { code: CODES.ERRORS.ERR__REVOKE_PARTITIONS }, + [{ topic: 'my-test-topic', partition: 0 }]); + + queueIdle = true; + ledgerCount = 0; + drainCallbacks[drainCallbacks.length - 1](); + + setImmediate(() => { + assert.strictEqual(closed, true); + done(); + }); + }); + }); + + it('should leave the group anyway when the drain never completes', done => { + const clock = sinon.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + queueIdle = false; + ledgerCount = 1; + + let closed = false; + consumer.close(() => { + closed = true; + }); + + setImmediate(() => { + assert.strictEqual(closed, false); + clock.tick(consumer._maxPollIntervalMs); + + setImmediate(() => { + clock.restore(); + assert.strictEqual(closed, true); + assert(consumer._consumer.unassign.calledOnce); + assert(consumer._consumer.unsubscribe.calledOnce); + done(); + }); + }); + }); + + it('should leave the group when only the ledger is stuck', done => { + // the replication queue processor's shape: its stop() closes the + // status producer first, and a task's offset is only committable + // from that producer's delivery callback, so a report that never + // arrives leaves a ledger entry outstanding with the queue idle + const clock = sinon.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + queueIdle = true; + ledgerCount = 1; + + let closed = false; + consumer.close(() => { + closed = true; + }); + + setImmediate(() => { + assert.strictEqual(closed, false); + clock.tick(consumer._maxPollIntervalMs); + + setImmediate(() => { + clock.restore(); + assert.strictEqual(closed, true); + assert(consumer._consumer.unassign.calledOnce); + done(); + }); + }); + }); + + 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 + queueIdle = false; + ledgerCount = 1; + consumer._consumer.isConnected = () => false; + + consumer.close(() => { + assert(consumer._consumer.disconnect.notCalled); + done(); + }); + }); + + it('should not wait for an in-flight offset publish', done => { + consumer._publishOffsetsCronActive = true; + consumer.close(() => { + assert(consumer._consumer.disconnect.calledOnce); + done(); + }); + }); + + it('should stop waiting for a disconnect that never completes', done => { + // setImmediate has to stay real, the test drives the flow with it + const clock = sinon.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + consumer._consumer.once = () => {}; + + let closed = false; + consumer.close(() => { + closed = true; + }); + + setImmediate(() => { + clock.tick(10000); + setImmediate(() => { + clock.restore(); + assert.strictEqual(closed, true); + assert(consumer._consumer.unsubscribe.calledOnce); + done(); + }); + }); + }); + + it('should drop a rebalance watchdog armed before the shutdown', done => { + consumer._drainProcessQueueTimeout = setTimeout(() => { + done(new Error('the rebalance watchdog fired during shutdown')); + }, 20); + + consumer.close(() => { + assert.strictEqual(consumer._drainProcessQueueTimeout, null); + setTimeout(done, 40); + }); + }); + }); });