diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index 9cb93693a..8a439403a 100644 --- a/lib/BackbeatConsumer.js +++ b/lib/BackbeatConsumer.js @@ -206,6 +206,10 @@ class BackbeatConsumer extends EventEmitter { this._disconnecting = false; // True between receiving a revoke and answering it (draining) this._waitingDrain = false; + // Set when a grant lands on a revoke we have not answered yet: + // the group has moved on without us, so we stop consuming and + // leave once the pending drain has committed + this._abortingRebalance = false; // Bounds close()'s wait for the drain this._closeDrainTimeout = null; @@ -445,10 +449,10 @@ class BackbeatConsumer extends EventEmitter { this._tryConsumedTimeout = null; } - // stop fetching once closing: new entries would refill the - // pipeline close() is draining, and this ends the + // stop fetching once closing or aborting: new entries would + // refill the pipeline being drained, and this ends the // self-rescheduling consume loop that would poll a closed client - if (this._closing) { + if (this._closing || this._abortingRebalance) { return undefined; } @@ -474,6 +478,20 @@ class BackbeatConsumer extends EventEmitter { this._nConsumePendingRequests += nNewConsumeRequests; return this._consumer.consume(nNewConsumeRequests, (err, entries) => { this._nConsumePendingRequests -= nNewConsumeRequests; + if (this._abortingRebalance) { + // A request outstanding when the abort started can still + // deliver, and the grant we accepted reset the fetch + // position to the last commit, so these are the entries + // being drained coming back around. Drop them: they are + // uncommitted, and whoever takes the partitions over + // reads them again from that same commit. + this._log.info('dropping entries consumed while aborting', { + topic: this._topic, + groupId: this._groupId, + count: entries ? entries.length : 0, + }); + return; + } if (!err) { entries.forEach(entry => { const { topic, partition, offset, key, timestamp } = entry; @@ -772,6 +790,76 @@ class BackbeatConsumer extends EventEmitter { } } + /** + * Handle a grant that lands on a revoke we have not answered yet. + * + * librdkafka answers a subscribed-topic metadata change with an + * immediate rejoin, without waiting for the un-assign the revoke + * asked for. The generation granted here has already moved past the + * one the draining entries belong to, so carrying on would mean + * consuming those partitions again from their last commit while + * still finishing the previous generation's work: the same entries + * processed twice, concurrently for consumers that do not order by + * key. Treat it like the drain timeout instead. Stop consuming, let + * the pending drain commit what it has, then leave the group so the + * partitions are taken over and this consumer is restarted. + * + * The grant is accepted rather than declined because the draining + * entries can only store their offsets while the partitions are + * held. + * + * @param {TopicPartition[]} assignment - partitions being granted + * @returns {undefined} + */ + _abortSupersededRebalance(assignment) { + // Keep answering grants while leaving, or the client stays parked + this._bestEffort('assign', () => this._consumer.assign(assignment)); + if (this._abortingRebalance) { + return; + } + this._log.error('rdkafka.assign while a revoke is still draining: ' + + 'the group moved on without us, aborting', { + topic: this._topic, + groupId: this._groupId, + assignment, + queueLen: this._processingQueue.length(), + running: this._processingQueue.running(), + ledger: this._offsetLedger.getProcessingCount(this._topic), + }); + KafkaBacklogMetrics.onRebalance(this._topic, this._groupId, + unassignStatus.SUPERSEDED); + this._abortingRebalance = true; + // The pending drain commits and un-assigns as it always does; + // once it has, there is nothing left to lose by leaving. + this.once('unassign', () => this._leaveAfterAbort().catch(err => { + this._log.error('failed to leave after an aborted rebalance', { + topic: this._topic, + groupId: this._groupId, + error: err.message, + }); + })); + } + + /** + * Leave the group and stop being ready after an aborted rebalance, + * so the liveness probe (or supervisord on S3C) restarts us. + * @returns {Promise} resolves once the client is closed + */ + async _leaveAfterAbort() { + await this._close(); + this._log.fatal('consumer left the queue after an aborted rebalance, ' + + 'restart needed', { + topic: this._topic, + groupId: this._groupId, + }); + if (process.env.CRASH_ON_REBALANCE_TIMEOUT === 'true') { + // On S3C nothing reacts to the healthcheck that fails once + // disconnected, so exit and let supervisord restart the + // program; grace period so the fatal line flushes first. + setTimeout(() => process.exit(1), 1000); + } + } + /** * @param {kafka.KafkaError} err Rebalance event * @param {TopicPartition[]} assignment List of (un)assigned partitions @@ -792,6 +880,10 @@ class BackbeatConsumer extends EventEmitter { } return; } + if (this._waitingDrain) { + this._abortSupersededRebalance(assignment); + return; + } this._log.info('rdkafka.assign', { assignment }); try { diff --git a/lib/constants.js b/lib/constants.js index 6aa5c2f71..404dd6eb8 100644 --- a/lib/constants.js +++ b/lib/constants.js @@ -26,6 +26,7 @@ const constants = { DRAINED: 'drained', TIMEOUT: 'timeout', SHUTDOWN: 'shutdown', + SUPERSEDED: 'superseded', }, statusReady: 'READY', statusUndefined: 'UNDEFINED', diff --git a/tests/functional/lib/BackbeatConsumerRebalanceAbort.js b/tests/functional/lib/BackbeatConsumerRebalanceAbort.js new file mode 100644 index 000000000..626077911 --- /dev/null +++ b/tests/functional/lib/BackbeatConsumerRebalanceAbort.js @@ -0,0 +1,264 @@ +const assert = require('assert'); +const { promisify } = require('util'); +const kafka = require('node-rdkafka'); +const sinon = require('sinon'); + +const BackbeatProducer = require('../../../lib/BackbeatProducer'); +const BackbeatConsumer = require('../../../lib/BackbeatConsumer'); +const { withTopicPrefix } = require('../../../lib/util/topic'); +const { unassignStatus } = require('../../../lib/constants'); + +const zookeeperConf = { connectionString: 'localhost:2181' }; +const kafkaConf = { hosts: 'localhost:9092' }; + +const { ERR__ASSIGN_PARTITIONS, ERR__REVOKE_PARTITIONS } = kafka.CODES.ERRORS; + +const sleep = ms => new Promise(resolve => setTimeout(resolve, ms)); + +function deferred() { + let resolve; + const promise = new Promise(res => { + resolve = res; + }); + return { promise, resolve }; +} + +class InstrumentedConsumer extends BackbeatConsumer { + _onRebalance(err, assignment) { + if (!this.rebalanceLog) { + this.rebalanceLog = []; + } + this.rebalanceLog.push(err.code); + super._onRebalance(err, assignment); + } +} + +// When a revoke arrives with work in flight, the un-assign is deferred +// until the drain completes. librdkafka however answers a subscribed-topic +// metadata change with an immediate rejoin even while that revoke is +// still unanswered, so a fresh assignment can be granted for a generation +// that has already moved past the work still draining. +// +// The consumer must then stop consuming, let the drain commit what it has +// and leave the group, rather than carry on and reprocess those entries +// alongside their own redelivery. +// +// The trigger is manufactured with partition-count bumps, observed by an +// application getMetadata() call. If librdkafka ever stops rejoining while +// a revoke is unanswered, the setup fails on 'no assign delivered' rather +// than on the assertions that follow. +describe('BackbeatConsumer aborts a superseded rebalance', function testSuite() { + this.timeout(120000); + + let admin; + let producer; + let consumer; + let rawTopic; + let fullTopic; + let groupId; + let taskStarted; + let taskGate; + let unassigns; + let processed; + + function queueProcessor(message, cb) { + const value = message.value.toString(); + processed.push(value); + if (value === 'hold') { + taskStarted.resolve(); + taskGate.promise.then(() => cb()); + return; + } + process.nextTick(cb); + } + + beforeEach(() => { + rawTopic = `backbeat-abort-spec-${Date.now()}`; + fullTopic = withTopicPrefix(rawTopic); + groupId = `abort-group-${Math.random()}`; + taskStarted = deferred(); + taskGate = deferred(); + unassigns = []; + processed = []; + admin = kafka.AdminClient.create({ + 'client.id': 'abort-spec-admin', + 'metadata.broker.list': kafkaConf.hosts, + }); + }); + + afterEach(async function teardown() { + this.timeout(40000); + sinon.restore(); + taskGate.resolve(); + if (consumer) { + await Promise.race([ + promisify(consumer.close.bind(consumer))(), + sleep(15000), + ]); + } + if (producer) { + await promisify(producer.close.bind(producer))(); + } + try { + admin.disconnect(); + } catch { + // already disconnected + } + consumer = null; + producer = null; + }); + + async function start() { + await promisify(admin.createTopic.bind(admin))({ + topic: fullTopic, + /* eslint-disable camelcase */ + num_partitions: 1, + replication_factor: 1, + /* eslint-enable camelcase */ + }, 15000); + + consumer = new InstrumentedConsumer({ + clientId: 'BackbeatConsumer-abort', + zookeeper: zookeeperConf, + kafka: kafkaConf, + groupId, + topic: rawTopic, + queueProcessor, + fromOffset: 'earliest', + // rebalance callbacks are delivered by the consume poll, + // which stops while the pipeline is full: a free slot must + // remain next to the held task for the revoke to reach us + concurrency: 2, + }); + consumer.on('unassign', status => unassigns.push(status)); + + producer = new BackbeatProducer({ + kafka: kafkaConf, + topic: rawTopic, + pollIntervalMs: 100, + }); + await Promise.all([ + new Promise(resolve => consumer.on('ready', resolve)), + new Promise(resolve => producer.on('ready', resolve)), + ]); + consumer.subscribe(); + } + + const send = messages => + promisify(producer.send.bind(producer))(messages); + + const bump = partitions => + promisify(admin.createPartitions.bind(admin))( + fullTopic, partitions, 15000); + + const sawRevoke = () => + (consumer.rebalanceLog || []).includes(ERR__REVOKE_PARTITIONS); + + const sawAssignAfterRevoke = () => { + const log = consumer.rebalanceLog || []; + const revokeIdx = log.indexOf(ERR__REVOKE_PARTITIONS); + return revokeIdx !== -1 && + log.slice(revokeIdx + 1).includes(ERR__ASSIGN_PARTITIONS); + }; + + // wait for cond, issuing the same full-cluster metadata request + // BackbeatConsumer itself makes, which librdkafka answers with a + // consumer-group subscription re-check + async function driveUntil(cond, what) { + for (let i = 0; i < 30; i++) { + if (cond()) { + return; + } + await promisify(consumer.getMetadata.bind(consumer))( + { allTopics: true, timeout: 10000 }).catch(() => {}); + await sleep(500); + } + assert.fail(what); + } + + async function waitUntil(cond, timeoutMs, what) { + const deadline = Date.now() + timeoutMs; + while (!cond()) { + if (Date.now() > deadline) { + assert.fail(typeof what === 'function' ? what() : what); + } + await sleep(100); + } + } + + // read the group's committed offset without joining the group: + // assign() alone does not make this client a member + async function committedOffset(partition) { + const probe = new kafka.KafkaConsumer({ + 'metadata.broker.list': kafkaConf.hosts, + 'group.id': groupId, + 'enable.auto.commit': false, + }, {}); + try { + await new Promise((resolve, reject) => { + probe.connect({ timeout: 10000 }, err => + (err ? reject(err) : resolve())); + }); + probe.assign([{ topic: fullTopic, partition }]); + const toppars = await promisify( + probe.committed.bind(probe))([{ topic: fullTopic, partition }], + 10000); + return toppars[0].offset; + } finally { + probe.disconnect(); + } + } + + it('leaves the group instead of consuming a generation it cannot ' + + 'commit for', async () => { + await start(); + + // hold one task in flight so the next revoke defers its un-assign + await send([{ key: 'k-hold', message: 'hold' }]); + await taskStarted.promise; + + // first bump: the next group-updating metadata response delivers + // a normal revoke, which parks behind the held task + await bump(2); + await driveUntil(sawRevoke, + 'no revoke delivered after the first partition bump'); + await sleep(300); + assert.deepStrictEqual(unassigns, [], + 'the un-assign should be deferred while a task is in flight'); + + // second bump while the revoke is unanswered: librdkafka rejoins + // and grants a fresh assignment before the drain completes + await bump(3); + await driveUntil(sawAssignAfterRevoke, + 'no assign delivered while the revoke was parked: the ' + + 'trigger is not reproducible on this librdkafka version'); + assert.strictEqual(consumer._abortingRebalance, true, + 'the superseded grant should have started an abort'); + + // the held task completes, so the drain commits and un-assigns + taskGate.resolve(); + await waitUntil(() => unassigns.length > 0, 20000, + 'the drain did not complete after the held task finished'); + // the drain ran to completion first; the departure that follows + // emits its own shutdown un-assign + assert.strictEqual(unassigns[0], unassignStatus.DRAINED, + `expected the drain to complete, got ${unassigns.join()}`); + + // ... and the consumer then leaves and reports unhealthy, so the + // liveness probe restarts it + await waitUntil(() => !consumer.isReady(), 20000, + 'the consumer stayed ready after aborting the rebalance'); + + // the drained work was committed, so whoever takes the partition + // over does not reprocess it + const committed = await committedOffset(0); + assert.strictEqual(committed, 1, + `expected the held entry to be committed, got ${committed}`); + + // and nothing was consumed twice, nor picked up after leaving + await send([{ key: 'after', message: 'after-leaving' }]); + await sleep(3000); + assert.deepStrictEqual(processed, ['hold'], + `expected only the held entry, got ${JSON.stringify(processed)}`); + }); +}); diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js index f97f7c635..b53f17ea5 100644 --- a/tests/unit/backbeatConsumer.js +++ b/tests/unit/backbeatConsumer.js @@ -3,6 +3,7 @@ const sinon = require('sinon'); const { EventEmitter } = require('events'); const BackbeatConsumer = require('../../lib/BackbeatConsumer'); +const KafkaBacklogMetrics = require('../../lib/KafkaBacklogMetrics'); const { CODES } = require('node-rdkafka'); const { kafka } = require('../config.json'); @@ -615,4 +616,167 @@ describe('backbeatConsumer', () => { }); }); }); + + describe('superseded rebalance', () => { + const REVOKE = { code: CODES.ERRORS.ERR__REVOKE_PARTITIONS }; + const ASSIGN = { code: CODES.ERRORS.ERR__ASSIGN_PARTITIONS }; + const partitions = [ + { topic: 'my-test-topic', partition: 0 }, + { topic: 'my-test-topic', partition: 1 }, + ]; + + let consumer; + let mockConsumer; + let queueIdle; + let ledgerCount; + + beforeEach(() => { + consumer = new BackbeatConsumerMock({ + kafka, + groupId: 'unittest-group', + topic: 'my-test-topic', + // a free slot must remain next to in-flight work, or the + // consume loop stops on its own + concurrency: 2, + }); + const mock = new EventEmitter(); + mockConsumer = Object.assign(mock, { + isConnected: () => true, + subscription: () => ['my-test-topic'], + assignments: () => [], + unsubscribe: sinon.stub(), + unassign: sinon.stub(), + assign: sinon.stub(), + commit: sinon.stub(), + pause: sinon.stub(), + resume: sinon.stub(), + consume: sinon.stub(), + disconnect: sinon.stub().callsFake(() => process.nextTick( + () => mock.emit('disconnected'))), + }); + consumer._consumer = mockConsumer; + + queueIdle = false; + ledgerCount = 1; + consumer._processingQueue = { + length: () => 0, + running: () => (queueIdle ? 0 : 1), + idle: () => queueIdle, + setDrain: () => {}, + push: sinon.stub(), + }; + consumer._offsetLedger.getProcessingCount = () => ledgerCount; + + sinon.stub(KafkaBacklogMetrics, 'onRebalance'); + }); + + afterEach(() => { + clearTimeout(consumer._drainProcessQueueTimeout); + sinon.restore(); + }); + + const completeDrain = () => { + queueIdle = true; + ledgerCount = 0; + if (consumer._drainCallback) { + consumer._drainCallback(); + } + }; + + it('should accept the grant and start aborting when it lands on a ' + + 'parked revoke', () => { + consumer._onRebalance(REVOKE, partitions); + assert.strictEqual(consumer._waitingDrain, true); + assert(mockConsumer.unassign.notCalled); + + consumer._onRebalance(ASSIGN, partitions); + + // accepted, so the draining entries can still store offsets + assert(mockConsumer.assign.calledOnceWithExactly(partitions)); + assert.strictEqual(consumer._abortingRebalance, true); + assert(KafkaBacklogMetrics.onRebalance.calledWith( + 'my-test-topic', 'unittest-group', + unassignStatus.SUPERSEDED)); + }); + + it('should stop consuming once aborting', () => { + consumer._onRebalance(REVOKE, partitions); + consumer._onRebalance(ASSIGN, partitions); + + consumer._tryConsume(); + assert(mockConsumer.consume.notCalled); + }); + + it('should drop entries delivered by a request outstanding when the ' + + 'abort started', () => { + const onOffsetConsumed = + sinon.spy(consumer._offsetLedger, 'onOffsetConsumed'); + mockConsumer.consume = sinon.stub().yields(null, [ + { topic: 'my-test-topic', partition: 0, offset: 0, + value: Buffer.from('a') }, + ]); + consumer._tryConsume(); + assert(onOffsetConsumed.calledOnce); + assert(consumer._processingQueue.push.calledOnce); + + consumer._onRebalance(REVOKE, partitions); + consumer._onRebalance(ASSIGN, partitions); + + // replay the callback of the request that was already + // outstanding: the grant reset the fetch position, so this is + // the draining entry coming back around + mockConsumer.consume.yield(null, [ + { topic: 'my-test-topic', partition: 0, offset: 0, + value: Buffer.from('a') }, + ]); + + assert(onOffsetConsumed.calledOnce); + assert(consumer._processingQueue.push.calledOnce); + }); + + it('should leave the group once the pending drain has committed', + done => { + consumer._onRebalance(REVOKE, partitions); + consumer._onRebalance(ASSIGN, partitions); + assert(mockConsumer.disconnect.notCalled); + + mockConsumer.disconnect = sinon.stub().callsFake(() => { + // the drain committed and un-assigned before we left + assert(mockConsumer.commit.calledOnce); + assert(mockConsumer.unassign.calledOnce); + assert(mockConsumer.unsubscribe.calledOnce); + process.nextTick(() => mockConsumer.emit('disconnected')); + done(); + }); + completeDrain(); + }); + + it('should not abort on a grant that follows a completed revoke', + () => { + queueIdle = true; + ledgerCount = 0; + consumer._onRebalance(REVOKE, partitions); + assert.strictEqual(consumer._waitingDrain, false); + + consumer._onRebalance(ASSIGN, partitions); + + assert.strictEqual(consumer._abortingRebalance, false); + assert(mockConsumer.assign.calledOnceWithExactly(partitions)); + assert(KafkaBacklogMetrics.onRebalance.neverCalledWith( + 'my-test-topic', 'unittest-group', + unassignStatus.SUPERSEDED)); + }); + + it('should decline the grant rather than abort when already ' + + 'shutting down', () => { + consumer._onRebalance(REVOKE, partitions); + consumer._closing = true; + + consumer._onRebalance(ASSIGN, partitions); + + assert.strictEqual(consumer._abortingRebalance, false); + assert(mockConsumer.assign.notCalled); + assert(mockConsumer.unassign.calledOnce); + }); + }); });