oplog: Use local _liveBuckets instead of unnecessary pipeline fetch

Issue: BB-491

Author: delthas

extensions/oplogPopulator/modules/Connector.js

@@ -45,6 +45,7 @@ class Connector extends EventEmitter {
         this._name = params.name;
         this._config = params.config;
         this._buckets = new Set(params.buckets);
+        this._liveBuckets = new Set(params.buckets);
         this._isRunning = params.isRunning;
         this._getPipeline = params.getPipeline;
         this._state = {
@@ -79,6 +80,12 @@ class Connector extends EventEmitter {
      */
     get buckets() { return [...this._buckets]; }
 
+    /**
+     * Getter for connector live buckets (currently configured on Kafka Connect)
+     * @returns {string[]} live buckets
+     */
+    get liveBuckets() { return [...this._liveBuckets]; }
+
     /**
      * Get number of buckets assigned to this
      * connector
@@ -166,6 +173,7 @@ class Connector extends EventEmitter {
                 config: this._config,
             });
             this._isRunning = true;
+            this._liveBuckets = new Set(this._buckets);
         } catch (err) {
             this._logger.error('Error while spawning connector', {
                 method: 'Connector.spawn',
@@ -193,6 +201,7 @@ class Connector extends EventEmitter {
             this.emit(constants.connectorUpdatedEvent, this);
             await this._kafkaConnect.deleteConnector(this._name);
             this._isRunning = false;
+            this._liveBuckets = new Set();
             // resetting the resume point to set a new one on creation of the connector
             delete this._config['startup.mode.timestamp.start.at.operation.time'];
         } catch (err) {
@@ -330,13 +339,10 @@ class Connector extends EventEmitter {
         try {
             if (doUpdate && this._isRunning) {
                 if (!updateSupported) {
-                    const currentConfig = await this._kafkaConnect.getConnectorConfig(this._name);
-                    const currentPipeline = currentConfig.pipeline
-                        ? JSON.parse(currentConfig.pipeline) : [];
-                    const newPipeline = JSON.parse(this._config.pipeline);
-                    const currentMatch = JSON.stringify(currentPipeline[0]?.$match);
-                    const newMatch = JSON.stringify(newPipeline[0]?.$match);
-                    if (currentMatch !== newMatch) {
+                    const sameSize = this._liveBuckets.size === this._buckets.size;
+                    const sameContent = sameSize &&
+                        [...this._liveBuckets].every(b => this._buckets.has(b));
+                    if (!sameContent) {
                         this._logger.info('Skipping pipeline update: match stage differs', {
                             method: 'Connector.updatePipeline',
                             connector: this._name,
@@ -348,6 +354,7 @@ class Connector extends EventEmitter {
                 this._state.isUpdating = true;
                 this.emit(constants.connectorUpdatedEvent, this);
                 await this._kafkaConnect.updateConnectorConfig(this._name, this._config);
+                this._liveBuckets = new Set(this._buckets);
                 this._updateConnectorState(false, timeBeforeUpdate);
                 this._state.isUpdating = false;
                 return true;

tests/unit/oplogPopulator/Connector.js

@@ -62,6 +62,12 @@ describe('Connector', () => {
             await connector.spawn();
             assert.notStrictEqual(partitionName, connector.config['offset.partition.name']);
         });
+        it('should set liveBuckets to current buckets on spawn', async () => {
+            sinon.stub(connector._kafkaConnect, 'createConnector').resolves();
+            connector._buckets = new Set(['bucket1', 'bucket2']);
+            await connector.spawn();
+            assert.deepStrictEqual(connector.liveBuckets.sort(), ['bucket1', 'bucket2']);
+        });
         it('should not try spawning a new connector when on is already existent', async () => {
             const createStub = sinon.stub(connector._kafkaConnect, 'createConnector')
                 .resolves();
@@ -81,6 +87,13 @@ describe('Connector', () => {
             assert(deleteStub.calledOnceWith('example-connector'));
             assert.strictEqual(connector.isRunning, false);
         });
+        it('should clear liveBuckets on destroy', async () => {
+            sinon.stub(connector._kafkaConnect, 'deleteConnector').resolves();
+            connector._isRunning = true;
+            connector._liveBuckets = new Set(['bucket1']);
+            await connector.destroy();
+            assert.deepStrictEqual(connector.liveBuckets, []);
+        });
         it('should not try destroying a new connector when connector is already destroyed', async () => {
             const deleteStub = sinon.stub(connector._kafkaConnect, 'deleteConnector')
                 .resolves();
@@ -258,42 +271,58 @@ describe('Connector', () => {
             assert(updateStub.notCalled);
         });
 
-        it('should update when updateSupported is false and match stages are the same', async () => {
-            const matchStage = { $match: { 'ns.coll': { $in: ['bucket1'] } } };
-            const newPipeline = JSON.stringify([matchStage, { $set: { 'fullDocument.value.location': {} } }]);
+        it('should update when updateSupported is false and live buckets match', async () => {
+            connector._buckets = new Set(['bucket1']);
+            connector._liveBuckets = new Set(['bucket1']);
             connector._state.bucketsGotModified = true;
             connector._state.isUpdating = false;
             connector._isRunning = true;
-            sinon.stub(connector, '_getPipeline').returns(newPipeline);
-            sinon.stub(connector._kafkaConnect, 'getConnectorConfig').resolves({
-                pipeline: JSON.stringify([matchStage]),
-            });
+            sinon.stub(connector, '_getPipeline').returns('example-pipeline');
             const updateStub = sinon.stub(connector._kafkaConnect, 'updateConnectorConfig')
                 .resolves();
             const didUpdate = await connector.updatePipeline(true, false);
             assert.strictEqual(didUpdate, true);
             assert(updateStub.calledOnce);
         });
 
-        it('should skip update when updateSupported is false and match stages differ', async () => {
-            const newPipeline = JSON.stringify([
-                { $match: { 'ns.coll': { $in: ['bucket1'] } } },
-            ]);
+        it('should skip update when updateSupported is false and live buckets differ', async () => {
+            connector._buckets = new Set(['bucket1']);
+            connector._liveBuckets = new Set(['different-bucket']);
             connector._state.bucketsGotModified = true;
             connector._state.isUpdating = false;
             connector._isRunning = true;
-            sinon.stub(connector, '_getPipeline').returns(newPipeline);
-            sinon.stub(connector._kafkaConnect, 'getConnectorConfig').resolves({
-                pipeline: JSON.stringify([
-                    { $match: { 'ns.coll': { $in: ['different-bucket'] } } },
-                ]),
-            });
+            sinon.stub(connector, '_getPipeline').returns('example-pipeline');
             const updateStub = sinon.stub(connector._kafkaConnect, 'updateConnectorConfig')
                 .resolves();
             const didUpdate = await connector.updatePipeline(true, false);
             assert.strictEqual(didUpdate, false);
             assert(updateStub.notCalled);
         });
+
+        it('should sync liveBuckets after successful update', async () => {
+            connector._buckets = new Set(['bucket1', 'bucket2']);
+            connector._liveBuckets = new Set();
+            connector._state.bucketsGotModified = true;
+            connector._state.isUpdating = false;
+            connector._isRunning = true;
+            sinon.stub(connector, '_getPipeline').returns('example-pipeline');
+            sinon.stub(connector._kafkaConnect, 'updateConnectorConfig').resolves();
+            await connector.updatePipeline(true);
+            assert.deepStrictEqual(connector.liveBuckets.sort(), ['bucket1', 'bucket2']);
+        });
+
+        it('should not sync liveBuckets after failed update', async () => {
+            connector._buckets = new Set(['bucket1']);
+            connector._liveBuckets = new Set();
+            connector._state.bucketsGotModified = true;
+            connector._state.isUpdating = false;
+            connector._isRunning = true;
+            sinon.stub(connector, '_getPipeline').returns('example-pipeline');
+            sinon.stub(connector._kafkaConnect, 'updateConnectorConfig')
+                .rejects(errors.InternalError);
+            await assert.rejects(() => connector.updatePipeline(true));
+            assert.deepStrictEqual(connector.liveBuckets, []);
+        });
     });
 
     describe('getConfigSizeInBytes', () => {