From 765dccc91d4ae9aeeb60b30bdb5f54a08c6d96d4 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Wed, 2 Sep 2026 16:26:44 +0200 Subject: [PATCH] Abort the rebalance when a grant supersedes a parked revoke On ERR__REVOKE_PARTITIONS the un-assign is deferred until in-flight work drains. librdkafka however answers a subscribed-topic metadata change with an immediate rejoin, without waiting for that un-assign, so a grant can arrive for a generation that has already moved past the work still draining. 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. The deferred un-assign then discarded the granted assignment outright, leaving a live group member owning partitions at the broker with no local assignment and nothing left to trigger a rebalance. Treat it like the drain timeout instead. Accept the grant, since the draining entries can only store their offsets while the partitions are held, stop consuming, let the pending drain commit and un-assign as it always does, then leave the group so the partitions are taken over and the liveness probe restarts this consumer. Entries delivered by a consume request that was already outstanding are dropped: the accepted grant reset the fetch position to the last commit, so those are the draining entries coming back around. Issue: BB-835 Claude-Session: https://claude.ai/code/session_01JQoM2qBo8JABC43pXSUiF8 --- lib/BackbeatConsumer.js | 98 ++++++- lib/constants.js | 1 + .../lib/BackbeatConsumerRebalanceAbort.js | 264 ++++++++++++++++++ tests/unit/backbeatConsumer.js | 164 +++++++++++ 4 files changed, 524 insertions(+), 3 deletions(-) create mode 100644 tests/functional/lib/BackbeatConsumerRebalanceAbort.js 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); + }); + }); });