From 2225ebb6bd6bca0af404992b0d3909eb6876862d Mon Sep 17 00:00:00 2001 From: Maha Benzekri Date: Tue, 18 Aug 2026 16:29:44 +0200 Subject: [PATCH 1/3] 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 66e9495d9..068db1710 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -619,10 +619,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 b0dac689ec67c69ebadcbac31e36183110d37627 Mon Sep 17 00:00:00 2001 From: Maha Benzekri Date: Tue, 25 Aug 2026 11:32:59 +0200 Subject: [PATCH 2/3] 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 | 223 +++++++++++++++++++++ 2 files changed, 238 insertions(+), 10 deletions(-) diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index 068db1710..df3864844 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -616,10 +616,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, @@ -628,12 +634,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); } @@ -735,6 +737,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 @@ -742,7 +750,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 @@ -753,8 +760,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..11d042f9e 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,200 @@ 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( + null, + 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); + assert.strictEqual(removeReaderState.firstCall.args[1], + `${ingestionConfig.zookeeperPath}/${ACTIVE_BUCKET}`); + }); + + 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 = { From 4a35f253a5b3cc267c0db098f5ef6f782a960b42 Mon Sep 17 00:00:00 2001 From: Maha Benzekri Date: Tue, 8 Sep 2026 09:09:57 +0200 Subject: [PATCH 3/3] Log when an in-flight reader is skipped after source removal Keep the setup mock in each test and drop the redundant interaction describe: the two methods are already covered separately. Issue: BB-846 --- lib/queuePopulator/IngestionPopulator.js | 5 + tests/unit/ingestion/IngestionPopulator.js | 176 ++++----------------- 2 files changed, 34 insertions(+), 147 deletions(-) diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index df3864844..7aa113a45 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -621,6 +621,11 @@ class IngestionPopulator { // its source may have been removed or replaced in the meantime. It // must then neither be activated nor retried. if (this._ingestionSources[zenkoBucket] !== logReader) { + this.log.info('skipping log reader after setup: source was removed or replaced', { + method: 'IngestionPopulator._setupUpdatedReaders', + zenkoBucket, + location: logReader.getLocationConstraint(), + }); return cb(); } if (err) { diff --git a/tests/unit/ingestion/IngestionPopulator.js b/tests/unit/ingestion/IngestionPopulator.js index 11d042f9e..b48258843 100644 --- a/tests/unit/ingestion/IngestionPopulator.js +++ b/tests/unit/ingestion/IngestionPopulator.js @@ -279,18 +279,11 @@ 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) { + function addLogReader(zenkoBucket) { const logReader = sinon.createStubInstance(IngestionReader); logReader.getTargetZenkoBucketName.returns(zenkoBucket); - logReader.setup.yieldsAsync(setupError); + ip._ingestionSources[zenkoBucket] = logReader; + ip.logReadersUpdate.push(logReader); return logReader; } @@ -300,38 +293,34 @@ describe('Ingestion Populator', () => { }); 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]; + const logReader = addLogReader('bucket1'); + logReader.setup.yieldsAsync(null); ip._setupUpdatedReaders(err => { assert.ifError(err); - assert.deepStrictEqual(ip.logReaders, [logReaderMock]); + assert.deepStrictEqual(ip.logReaders, [logReader]); 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]; + const logReader = addLogReader('bucket1'); + logReader.setup.yieldsAsync(errors.InternalError); ip._setupUpdatedReaders(err => { assert.ifError(err); assert.deepStrictEqual(ip.logReaders, []); - assert.deepStrictEqual(ip.logReadersUpdate, [logReaderMock]); + assert.deepStrictEqual(ip.logReadersUpdate, [logReader]); 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]; + const logReader = addLogReader('bucket1'); + logReader.setup.yieldsAsync(errors.InternalError); + delete ip._ingestionSources.bucket1; ip._setupUpdatedReaders(err => { assert.ifError(err); @@ -343,11 +332,11 @@ describe('Ingestion Populator', () => { 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]; + const staleReader = addLogReader('bucket1'); + staleReader.setup.yieldsAsync(errors.InternalError); + const currentReader = sinon.createStubInstance(IngestionReader); + currentReader.getTargetZenkoBucketName.returns('bucket1'); + ip._ingestionSources.bucket1 = currentReader; ip._setupUpdatedReaders(err => { assert.ifError(err); @@ -359,9 +348,9 @@ 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]; + const logReader = addLogReader('bucket1'); + logReader.setup.yieldsAsync(null); + delete ip._ingestionSources.bucket1; ip._setupUpdatedReaders(err => { assert.ifError(err); @@ -373,10 +362,11 @@ describe('Ingestion Populator', () => { 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]; + const staleReader = addLogReader('bucket1'); + staleReader.setup.yieldsAsync(null); + const currentReader = sinon.createStubInstance(IngestionReader); + currentReader.getTargetZenkoBucketName.returns('bucket1'); + ip._ingestionSources.bucket1 = currentReader; ip._setupUpdatedReaders(err => { assert.ifError(err); @@ -387,12 +377,10 @@ describe('Ingestion Populator', () => { }); 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]; + const failingReader = addLogReader('bucket1'); + failingReader.setup.yieldsAsync(errors.InternalError); + const workingReader = addLogReader('bucket2'); + workingReader.setup.yieldsAsync(null); ip._setupUpdatedReaders(err => { assert.ifError(err); @@ -403,112 +391,6 @@ 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';