From 2f9c7f5e295b19640e0e3e046c0d55e7c02fce84 Mon Sep 17 00:00:00 2001 From: Maha Benzekri Date: Tue, 18 Aug 2026 16:29:44 +0200 Subject: [PATCH 1/2] Retry the setup of an ingestion reader when it fails When a reader's setup fails (typically the getRaftId call to the source), it was dropped from the pending list and never set up again: the reader stays registered in _ingestionSources, so applyUpdates considers the bucket already handled and only refreshes it. The bucket was therefore left out of ingestion until the pod was restarted. Queue the reader again instead, so it is retried on the next cycle. This follows what _processLogReaderEntries already does with read failures, and keeps the other locations ingesting, which restarting the pod on a failed healthcheck would not. The reader is only queued again if it is still the one registered for its bucket, so a source removed or replaced in the meantime is not brought back. Issue: BB-846 --- lib/queuePopulator/IngestionPopulator.js | 13 ++- tests/unit/ingestion/IngestionPopulator.js | 96 ++++++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index ec2849a40..7d6af28db 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -616,10 +616,21 @@ class IngestionPopulator { if (err) { // if setup fails for a log reader, don't add it to `logReaders` // log the error and continue setting up others - this.log.fatal('error setting up log reader', { + const zenkoBucket = logReader.getTargetZenkoBucketName(); + this.log.error('error setting up log reader, retrying later', { method: 'IngestionPopulator._setupUpdatedReaders', + zenkoBucket, + location: logReader.getLocationConstraint(), error: err, }); + // Setup failures are usually transient (source unreachable, + // invalid credentials...): queue the reader again so that it + // is retried on the next cycle, unless it is no longer the + // reader registered for its bucket, meaning its source was + // removed or replaced in the meantime. + if (this._ingestionSources[zenkoBucket] === logReader) { + this.logReadersUpdate.push(logReader); + } } else { this.logReaders.push(logReader); } diff --git a/tests/unit/ingestion/IngestionPopulator.js b/tests/unit/ingestion/IngestionPopulator.js index ad10e1904..553022f10 100644 --- a/tests/unit/ingestion/IngestionPopulator.js +++ b/tests/unit/ingestion/IngestionPopulator.js @@ -278,6 +278,102 @@ describe('Ingestion Populator', () => { }); }); + describe('_setupUpdatedReaders', () => { + const FAILING_BUCKET = 'failing-zenko-bucket'; + const WORKING_BUCKET = 'working-zenko-bucket'; + + /** + * @param {string} zenkoBucket - target zenko bucket of the reader + * @param {Error|null} setupError - error to fail `setup` with + * @return {object} the stubbed reader + */ + function createLogReaderMock(zenkoBucket, setupError) { + const logReader = sinon.createStubInstance(IngestionReader); + logReader.getTargetZenkoBucketName.returns(zenkoBucket); + logReader.setup.yieldsAsync(setupError); + return logReader; + } + + beforeEach(() => { + ip.logReaders = []; + ip.logReadersUpdate = []; + }); + + it('should activate a log reader once its setup succeeds', done => { + const logReaderMock = createLogReaderMock(WORKING_BUCKET, null); + ip._ingestionSources[WORKING_BUCKET] = logReaderMock; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, [logReaderMock]); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should queue a log reader again when its setup fails', done => { + const logReaderMock = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + ip._ingestionSources[FAILING_BUCKET] = logReaderMock; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, [logReaderMock]); + done(); + }); + }); + + it('should not queue a log reader again when its setup fails and ' + + 'its source is no longer configured', done => { + const logReaderMock = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + delete ip._ingestionSources[FAILING_BUCKET]; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should not queue a log reader again when its setup fails and ' + + 'its source has been registered with another reader', done => { + const staleReader = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + const currentReader = createLogReaderMock(FAILING_BUCKET, null); + ip._ingestionSources[FAILING_BUCKET] = currentReader; + ip.logReadersUpdate = [staleReader]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should keep setting up other log readers when one fails', done => { + const failingReader = + createLogReaderMock(FAILING_BUCKET, errors.InternalError); + const workingReader = createLogReaderMock(WORKING_BUCKET, null); + ip._ingestionSources[FAILING_BUCKET] = failingReader; + ip._ingestionSources[WORKING_BUCKET] = workingReader; + ip.logReadersUpdate = [failingReader, workingReader]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, [workingReader]); + assert.deepStrictEqual(ip.logReadersUpdate, [failingReader]); + done(); + }); + }); + }); + describe('_processLogReaderEntries', () => { it('should skip when previous batch currently in progress', () => { const logReaderMock = { From 3d8f265d99e98763b7f5b8326e13327710a50670 Mon Sep 17 00:00:00 2001 From: Maha Benzekri Date: Tue, 25 Aug 2026 11:32:59 +0200 Subject: [PATCH 2/2] Unregister an ingestion source as soon as it is removed A reader whose setup is in flight is in neither `logReaders` nor `logReadersUpdate`, so `_closeLogState` found nothing to remove and left its `_ingestionSources` entry behind. The reader then registered itself once its setup completed: on failure it was queued again, and on success it became active and could process a batch on a location that no longer exists. Unregister the source unconditionally, which makes `_ingestionSources` authoritative, and hoist the guard above the error branch so that it covers the success path as well. Issue: BB-846 --- lib/queuePopulator/IngestionPopulator.js | 25 ++- tests/unit/ingestion/IngestionPopulator.js | 220 +++++++++++++++++++++ 2 files changed, 235 insertions(+), 10 deletions(-) diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index 7d6af28db..84cf17440 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -613,10 +613,16 @@ class IngestionPopulator { const newReaders = this.logReadersUpdate; this.logReadersUpdate = []; async.each(newReaders, (logReader, cb) => logReader.setup(err => { + const zenkoBucket = logReader.getTargetZenkoBucketName(); + // The reader is in neither list while its setup is in flight, so + // its source may have been removed or replaced in the meantime. It + // must then neither be activated nor retried. + if (this._ingestionSources[zenkoBucket] !== logReader) { + return cb(); + } if (err) { // if setup fails for a log reader, don't add it to `logReaders` // log the error and continue setting up others - const zenkoBucket = logReader.getTargetZenkoBucketName(); this.log.error('error setting up log reader, retrying later', { method: 'IngestionPopulator._setupUpdatedReaders', zenkoBucket, @@ -625,12 +631,8 @@ class IngestionPopulator { }); // Setup failures are usually transient (source unreachable, // invalid credentials...): queue the reader again so that it - // is retried on the next cycle, unless it is no longer the - // reader registered for its bucket, meaning its source was - // removed or replaced in the meantime. - if (this._ingestionSources[zenkoBucket] === logReader) { - this.logReadersUpdate.push(logReader); - } + // is retried on the next cycle. + this.logReadersUpdate.push(logReader); } else { this.logReaders.push(logReader); } @@ -732,6 +734,12 @@ class IngestionPopulator { * @return {undefined} */ _closeLogState(key) { + // Unregister the source first: a reader whose setup is in flight is in + // neither list, and leaving its entry here would let it register itself + // once the setup completes. + const reader = this._ingestionSources[key]; + delete this._ingestionSources[key]; + if (this._checkAndRemoveLogReader(key, this.logReadersUpdate)) { // if removed from `this.logReadersUpdate`, zookeeper setup has not // been performed yet @@ -739,7 +747,6 @@ class IngestionPopulator { method: 'IngestionPopulator._closeLogState', bucket: key, }); - delete this._ingestionSources[key]; } if (this._checkAndRemoveLogReader(key, this.logReaders)) { // if removed from `this.logReaders`, we must cleanup zookeeper @@ -750,8 +757,6 @@ class IngestionPopulator { }); // we should first validate this reader is not currently processing // a batch of entries before removing zookeeper state - const reader = this._ingestionSources[key]; - delete this._ingestionSources[key]; const path = `${this.ingestionConfig.zookeeperPath}/${key}`; this._removeReaderState(reader, path); } diff --git a/tests/unit/ingestion/IngestionPopulator.js b/tests/unit/ingestion/IngestionPopulator.js index 553022f10..9abb717cb 100644 --- a/tests/unit/ingestion/IngestionPopulator.js +++ b/tests/unit/ingestion/IngestionPopulator.js @@ -357,6 +357,35 @@ describe('Ingestion Populator', () => { }); }); + it('should not activate a log reader when its setup succeeds and ' + + 'its source is no longer configured', done => { + const logReaderMock = createLogReaderMock(WORKING_BUCKET, null); + delete ip._ingestionSources[WORKING_BUCKET]; + ip.logReadersUpdate = [logReaderMock]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + + it('should not activate a log reader when its setup succeeds and ' + + 'its source has been registered with another reader', done => { + const staleReader = createLogReaderMock(WORKING_BUCKET, null); + const currentReader = createLogReaderMock(WORKING_BUCKET, null); + ip._ingestionSources[WORKING_BUCKET] = currentReader; + ip.logReadersUpdate = [staleReader]; + + ip._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(ip.logReaders, []); + assert.deepStrictEqual(ip.logReadersUpdate, []); + done(); + }); + }); + it('should keep setting up other log readers when one fails', done => { const failingReader = createLogReaderMock(FAILING_BUCKET, errors.InternalError); @@ -374,6 +403,197 @@ describe('Ingestion Populator', () => { }); }); + describe('removing a source while its setup is in flight', () => { + const REMOVED_BUCKET = 'removed-zenko-bucket'; + + // `IngestionPopulatorMock` stubs out `_closeLogState`, so a real + // populator is needed to exercise the removal path. + let populator; + + beforeEach(() => { + populator = new IngestionPopulator( + zkConfig, + kafkaConfig, + qpConfig, + mConfig, + rConfig, + ingestionConfig, + s3Config, + ); + }); + + /** + * Build a reader whose `setup` stays pending until it is released, to + * hold the reader in neither `logReaders` nor `logReadersUpdate`. + * + * @param {string} zenkoBucket - target zenko bucket of the reader + * @param {Error|null} setupError - error to fail `setup` with + * @return {object} the stubbed reader and the release function + */ + function createPendingLogReaderMock(zenkoBucket, setupError) { + const logReader = sinon.createStubInstance(IngestionReader); + logReader.getTargetZenkoBucketName.returns(zenkoBucket); + + let setupCb = null; + logReader.setup.callsFake(cb => { + setupCb = cb; + }); + + return { logReader, release: () => setupCb(setupError) }; + } + + it('should not queue a log reader again when its source is removed ' + + 'while its setup fails', done => { + const { logReader, release } = + createPendingLogReaderMock(REMOVED_BUCKET, errors.InternalError); + populator._ingestionSources[REMOVED_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(populator.logReadersUpdate, []); + done(); + }); + + populator._closeLogState(REMOVED_BUCKET); + assert.strictEqual(populator._ingestionSources[REMOVED_BUCKET], + undefined); + + release(); + }); + + it('should not activate a log reader when its source is removed ' + + 'while its setup succeeds', done => { + const { logReader, release } = + createPendingLogReaderMock(REMOVED_BUCKET, null); + populator._ingestionSources[REMOVED_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(populator.logReaders, []); + assert.deepStrictEqual(populator.logReadersUpdate, []); + done(); + }); + + populator._closeLogState(REMOVED_BUCKET); + release(); + }); + + it('should not activate a log reader when its source is replaced ' + + 'while its setup succeeds', done => { + const { logReader, release } = + createPendingLogReaderMock(REMOVED_BUCKET, null); + const freshReader = sinon.createStubInstance(IngestionReader); + freshReader.getTargetZenkoBucketName.returns(REMOVED_BUCKET); + + populator._ingestionSources[REMOVED_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._setupUpdatedReaders(err => { + assert.ifError(err); + assert.deepStrictEqual(populator.logReaders, []); + assert.deepStrictEqual(populator.logReadersUpdate, []); + // the reader registered in the meantime is left untouched + assert.strictEqual(populator._ingestionSources[REMOVED_BUCKET], + freshReader); + done(); + }); + + // the source is removed, then configured again before the setup + // of the first reader completes + populator._closeLogState(REMOVED_BUCKET); + populator._ingestionSources[REMOVED_BUCKET] = freshReader; + + release(); + }); + }); + + describe('_closeLogState', () => { + const ACTIVE_BUCKET = 'active-zenko-bucket'; + const PENDING_BUCKET = 'pending-zenko-bucket'; + + // `IngestionPopulatorMock` stubs out `_closeLogState`, so a real + // populator is needed to exercise it. + let populator; + let removeReaderState; + + beforeEach(() => { + populator = new IngestionPopulator( + zkConfig, + kafkaConfig, + qpConfig, + mConfig, + rConfig, + ingestionConfig, + s3Config, + ); + // `_removeReaderState` polls the reader on a timer that would + // outlive the test + removeReaderState = sinon.stub(populator, '_removeReaderState'); + }); + + /** + * @param {string} zenkoBucket - target zenko bucket of the reader + * @return {object} the stubbed reader + */ + function createReaderMock(zenkoBucket) { + const logReader = sinon.createStubInstance(IngestionReader); + logReader.getTargetZenkoBucketName.returns(zenkoBucket); + return logReader; + } + + it('should unregister a source whose reader is still pending setup', + () => { + const logReader = createReaderMock(PENDING_BUCKET); + populator._ingestionSources[PENDING_BUCKET] = logReader; + populator.logReadersUpdate = [logReader]; + + populator._closeLogState(PENDING_BUCKET); + + assert.deepStrictEqual(populator.logReadersUpdate, []); + assert.strictEqual(populator._ingestionSources[PENDING_BUCKET], + undefined); + // zookeeper state is only created once the setup succeeded + assert.strictEqual(removeReaderState.called, false); + }); + + it('should unregister an active source and clean its zookeeper state', + () => { + const logReader = createReaderMock(ACTIVE_BUCKET); + populator._ingestionSources[ACTIVE_BUCKET] = logReader; + populator.logReaders = [logReader]; + + populator._closeLogState(ACTIVE_BUCKET); + + assert.deepStrictEqual(populator.logReaders, []); + assert.strictEqual(populator._ingestionSources[ACTIVE_BUCKET], + undefined); + // the reader is read before being unregistered, as + // `_removeReaderState` polls it until its batch completes + assert.strictEqual(removeReaderState.calledOnce, true); + assert.strictEqual(removeReaderState.firstCall.args[0], logReader); + }); + + it('should unregister a source whose setup is still in flight', () => { + const logReader = createReaderMock(PENDING_BUCKET); + populator._ingestionSources[PENDING_BUCKET] = logReader; + + populator._closeLogState(PENDING_BUCKET); + + assert.strictEqual(populator._ingestionSources[PENDING_BUCKET], + undefined); + assert.strictEqual(removeReaderState.called, false); + }); + + it('should not throw when the source is unknown', () => { + populator._closeLogState('never-configured-bucket'); + + assert.deepStrictEqual(populator._ingestionSources, {}); + assert.strictEqual(removeReaderState.called, false); + }); + }); + describe('_processLogReaderEntries', () => { it('should skip when previous batch currently in progress', () => { const logReaderMock = {