Ensure no race condition when handling connector events

Issue: BB-601

extensions/oplogPopulator/OplogPopulator.js

@@ -287,7 +287,7 @@ class OplogPopulator {
             });
             // For now, we always use the RetainBucketsDecorator
             // so, we map the events from the classes
-            this._connectorsManager.on('connector-updated', connector =>
+            this._connectorsManager.on(constants.connectorUpdatedEvent, connector =>
                 this._allocationStrategy.onConnectorUpdatedOrDestroyed(connector));
             this._allocator.on('bucket-removed', (bucket, connector) =>
                 this._allocationStrategy.onBucketRemoved(bucket, connector));

extensions/oplogPopulator/constants.js

@@ -6,6 +6,7 @@ const constants = {
     // to 16 MB (16777216B) / 64 (max length of a bucket name) ~= 260000
     maxBucketsPerConnector: 260000,
     mongodbVersionWithImmutablePipelines: '6.0.0',
+    connectorUpdatedEvent: 'connector-updated',
     defaultConnectorConfig: {
         'connector.class': 'com.mongodb.kafka.connect.MongoSourceConnector',
         'pipeline': '[]',

extensions/oplogPopulator/modules/Connector.js

@@ -1,7 +1,9 @@
 const joi = require('joi');
 const uuid = require('uuid');
 const { errors } = require('arsenal');
+const { EventEmitter } = require('stream');
 const KafkaConnectWrapper = require('../../../lib/wrappers/KafkaConnectWrapper');
+const constants = require('../constants');
 
 const connectorParams = joi.object({
     name: joi.string().required(),
@@ -21,7 +23,7 @@ const connectorParams = joi.object({
  * destroy and update the config of the connector when adding
  * or removing buckets from it
  */
-class Connector {
+class Connector extends EventEmitter {
 
     /**
      * @constructor
@@ -36,6 +38,7 @@ class Connector {
      * @param {number} params.kafkaConnectPort kafka connect port
      */
     constructor(params) {
+        super();
         joi.attempt(params, connectorParams);
         this._name = params.name;
         this._config = params.config;
@@ -185,6 +188,7 @@ class Connector {
         }
         try {
             await this._kafkaConnect.deleteConnector(this._name);
+            this.emit(constants.connectorUpdatedEvent, this);
             this._isRunning = false;
             // resetting the resume point to set a new one on creation of the connector
             delete this._config['startup.mode.timestamp.start.at.operation.time'];
@@ -339,6 +343,7 @@ class Connector {
             if (doUpdate && this._isRunning) {
                 const timeBeforeUpdate = Date.now();
                 this._state.isUpdating = true;
+                this.emit(constants.connectorUpdatedEvent, this);
                 await this._kafkaConnect.updateConnectorConfig(this._name, this._config);
                 this._updateConnectorState(false, timeBeforeUpdate);
                 this._state.isUpdating = false;

extensions/oplogPopulator/modules/ConnectorsManager.js

@@ -128,6 +128,9 @@ class ConnectorsManager extends EventEmitter {
             kafkaConnectHost: this._kafkaConnectHost,
             kafkaConnectPort: this._kafkaConnectPort,
         });
+        connector.on(constants.connectorUpdatedEvent, connector => {
+            return this.emit(constants.connectorUpdatedEvent, connector);
+        });
         this._connectors.push(connector);
         return connector;
     }
@@ -237,7 +240,7 @@ class ConnectorsManager extends EventEmitter {
     }
 
     /**
-     * Spawns a connector when buckets are configured for it and is not running,
+     * Spawns a connector when buckets aremit(e configured for it and is not running,
      * or destroys connector with no buckets configured
      * @param {Connector} connector connector instance
      * @returns {Promise<Boolean>} true if connector state changed
@@ -246,7 +249,6 @@ class ConnectorsManager extends EventEmitter {
     async _spawnOrDestroyConnector(connector) {
         try {
             if (connector.isRunning && connector.bucketCount === 0) {
-                this.emit('connector-updated', connector);
                 await connector.destroy();
                 this._metricsHandler.onConnectorDestroyed();
                 this._logger.info('Successfully destroyed a connector', {
@@ -263,11 +265,7 @@ class ConnectorsManager extends EventEmitter {
                 });
                 return true;
             } else if (connector.isRunning && this._allocationStrategy.canUpdate()) {
-                const isPipelineUpdated = connector.updatePipeline(true);
-                if (isPipelineUpdated) {
-                    this.emit('connector-updated', connector);
-                }
-                return isPipelineUpdated;
+                return connector.updatePipeline(true);
             }
 
             return false;

tests/unit/oplogPopulator/ConnectorsManager.js

@@ -11,6 +11,7 @@ const OplogPopulatorMetrics =
     require('../../../extensions/oplogPopulator/OplogPopulatorMetrics');
 const RetainBucketsDecorator = require('../../../extensions/oplogPopulator/allocationStrategy/RetainBucketsDecorator');
 const LeastFullConnector = require('../../../extensions/oplogPopulator/allocationStrategy/LeastFullConnector');
+const constants = require('../../../extensions/oplogPopulator/constants');
 
 const logger = new werelogs.Logger('ConnectorsManager');
 
@@ -238,9 +239,9 @@ describe('ConnectorsManager', () => {
             connector1._isRunning = true;
             connector1._state.bucketsGotModified = false;
             connector1._buckets = new Set();
-            const emitStub = sinon.stub(connectorsManager, 'emit');
+            const emitStub = sinon.stub(connector1, 'emit');
             await connectorsManager._spawnOrDestroyConnector(connector1);
-            assert(emitStub.calledOnceWith('connector-updated', connector1));
+            assert(emitStub.calledOnceWith(constants.connectorUpdatedEvent, connector1));
         });
 
         it('should spawn a non running connector when buckets are configured', async () => {

tests/unit/oplogPopulator/oplogPopulator.js

@@ -18,6 +18,7 @@ const RetainBucketsDecorator = require('../../../extensions/oplogPopulator/alloc
 const LeastFullConnector = require('../../../extensions/oplogPopulator/allocationStrategy/LeastFullConnector');
 const ImmutableConnector = require('../../../extensions/oplogPopulator/allocationStrategy/ImmutableConnector');
 const AllocationStrategy = require('../../../extensions/oplogPopulator/allocationStrategy/AllocationStrategy');
+const constants = require('../../../extensions/oplogPopulator/constants');
 
 const oplogPopulatorConfig = {
     topic: 'oplog',
@@ -209,7 +210,7 @@ describe('OplogPopulator', () => {
             assert(initializeConnectorsManagerStub.calledOnce);
             const onConnectorUpdatedOrDestroyedStub =
             sinon.stub(oplogPopulator._allocationStrategy, 'onConnectorUpdatedOrDestroyed');
-            oplogPopulator._connectorsManager.emit('connector-updated');
+            oplogPopulator._connectorsManager.emit(constants.connectorUpdatedEvent);
             assert(onConnectorUpdatedOrDestroyedStub.calledOnce);
         });