From dedd655660d3da0a1f208becd561a80066a66334 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Fri, 28 Aug 2026 18:04:11 +0200 Subject: [PATCH 1/3] Extract the drain predicate into a method The processing queue and the offset ledger both have to be drained before partitions are released, and the shutdown path needs the same test the revoke path already makes. Lift it out of the revoke closure so it can be reused rather than restated. No behaviour change: the same two call sites, the same predicate. Issue: BB-833 --- lib/BackbeatConsumer.js | 31 ++++++++++++++++++------------- 1 file changed, 18 insertions(+), 13 deletions(-) diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index e0fd50d8a..390ca9487 100644 --- a/lib/BackbeatConsumer.js +++ b/lib/BackbeatConsumer.js @@ -755,6 +755,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. @@ -883,18 +899,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 +910,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); } }); From 162170df97560c9b1a1bea265f3a0dc5e677c252 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Tue, 1 Sep 2026 11:14:45 +0200 Subject: [PATCH 2/3] Hoist the functional waitFor helper to module scope waitFor lived inside the fromOffset suite, and the group departure tests need the same poll-until-true wait. Lift it out so it can be reused rather than restated, and hand back a promise instead of a callback since the new tests are async. No behaviour change: the same predicate, the same 200ms poll and the same deadline, with the one existing call site adapted to the promise. Issue: BB-833 --- tests/functional/lib/BackbeatConsumer.js | 33 +++++++++++++----------- 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/tests/functional/lib/BackbeatConsumer.js b/tests/functional/lib/BackbeatConsumer.js index 060cb3aee..bf8195a5c 100644 --- a/tests/functional/lib/BackbeatConsumer.js +++ b/tests/functional/lib/BackbeatConsumer.js @@ -24,6 +24,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()}`; @@ -1286,20 +1302,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 +1371,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); }); }); From 22f3a61bc8a99d702cadfe176d59b756f33f960a Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Fri, 28 Aug 2026 18:04:13 +0200 Subject: [PATCH 3/3] Leave the consumer group explicitly on shutdown close() unsubscribed and then waited for the rebalance callback to un-assign before disconnecting. librdkafka delivers no such callback when the consumer holds no assignment, and postpones the unsubscribe outright while a rebalance is in progress, so close() never returned and the pod was SIGKILLed with the member still registered at the broker. The group then held zero partitions until session.timeout.ms evicted it, which during a rolling update happens by construction: the new pod joins before the old one is told to stop. Release the partitions and drop the subscription before closing, so the close path has nothing to hand back to us: before unsubscribe -> wait for a revoke -> [drain, commit, unassign] -> disconnect after drain -> commit -> unsubscribe -> unassign -> disconnect The bracketed steps only ran if a revoke arrived. The order of the last two matters, and the mechanism is a flag rather than the assignment list: only unsubscribe() sets F_LEAVE_ON_UNASSIGN_DONE, and only unassign_done() consults it to send the LeaveGroup. Un-assigning first would clear the assignment with no state change and the LeaveGroup would never be armed. In-flight work is still drained first, so offsets are committed exactly as before, bounded as the revoke path already bounded it. That bound is inherited rather than chosen -- BB-854 shortens it. Issue: BB-833 --- lib/BackbeatConsumer.js | 129 ++++++++-- lib/constants.js | 1 + tests/functional/lib/BackbeatConsumer.js | 185 ++++++++++++++ tests/unit/backbeatConsumer.js | 294 +++++++++++++++++++++++ 4 files changed, 585 insertions(+), 24 deletions(-) diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index 390ca9487..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 @@ -808,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); @@ -828,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 => { @@ -1292,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 bf8195a5c..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; @@ -1291,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; 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); + }); + }); + }); });