Check if GC config exists before creating gc producer in replication

extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js

@@ -204,14 +204,16 @@ class ReplicationStatusProcessor {
      *   in PEM format
      * @param {Object} mConfig - metrics config
      * @param {String} mConfig.topic - metrics config kafka topic
+     * @param {Object} gcConfig - config for garbage collector
      */
     constructor(kafkaConfig, sourceConfig, repConfig,
-                internalHttpsConfig, mConfig) {
+                internalHttpsConfig, mConfig, gcConfig) {
         this.kafkaConfig = kafkaConfig;
         this.sourceConfig = sourceConfig;
         this.repConfig = repConfig;
         this.internalHttpsConfig = internalHttpsConfig;
         this.mConfig = mConfig;
+        this.gcConfig = gcConfig;
         this._consumer = null;
         this._gcProducer = null;
         this._mProducer = null;
@@ -370,8 +372,12 @@ class ReplicationStatusProcessor {
                 });
             },
             done => {
-                this._gcProducer = new GarbageCollectorProducer();
-                this._gcProducer.setupProducer(done);
+                if (this.gcConfig) {
+                    this._gcProducer = new GarbageCollectorProducer();
+                    this._gcProducer.setupProducer(done);
+                } else {
+                    done();
+                }
             },
             done => {
                 this._mProducer = new MetricsProducer(this.kafkaConfig,

extensions/replication/replicationStatusProcessor/task.js

@@ -13,11 +13,12 @@
 const sourceConfig = repConfig.source;
 const internalHttpsConfig = config.internalHttps;
 const mConfig = config.metrics;
+const gcConfig = config.extensions.gc;
 
 const { initManagement } = require('../../../lib/management/index');
 
 const replicationStatusProcessor = new ReplicationStatusProcessor(
-    kafkaConfig, sourceConfig, repConfig, internalHttpsConfig, mConfig);
+    kafkaConfig, sourceConfig, repConfig, internalHttpsConfig, mConfig, gcConfig);
 
 werelogs.configure({ level: config.log.logLevel,
      dump: config.log.dumpLevel });

tests/functional/replication/queueProcessor.js

@@ -18,6 +18,8 @@ const replicationConstants = require('../../../extensions/replication/constants'
 const ReplicationAPI =
       require('../../../extensions/replication/ReplicationAPI');
 const TestConfigurator = require('../../utils/TestConfigurator');
+const ReplicationStatusProcessor = require('../../../extensions/replication' +
+    '/replicationStatusProcessor/ReplicationStatusProcessor');
 
 /* eslint-disable max-len */
 
@@ -760,8 +762,6 @@ describe('queue processor functional tests with mocking', () => {
     let copyLocationResultsConsumer;
 
     const QueueProcessor = require('../../../extensions/replication/queueProcessor/QueueProcessor');
-    const ReplicationStatusProcessor = require('../../../extensions/replication' +
-                  '/replicationStatusProcessor/ReplicationStatusProcessor');
 
     before(function before(done) {
         this.timeout(60000);
@@ -1449,3 +1449,64 @@ describe('queue processor functional tests with mocking', () => {
         });
     });
 });
+
+describe('GC should be created if config is provided', () => {
+    it('should create a GC if config is provided', done => {
+        const replicationStatusProcessor = new ReplicationStatusProcessor(
+            { hosts: 'localhost:9092' },
+            { auth: { type: 'role',
+                      vault: { host: constants.source.vault,
+                               port: 7777 } },
+              s3: { host: constants.source.s3,
+                    port: 7777 },
+              transport: 'http',
+            },
+            { replicationStatusTopic:
+              'backbeat-func-test-repstatus',
+              replicationStatusProcessor: {
+                  retry: {
+                      timeoutS: 5,
+                  },
+                  groupId: 'backbeat-func-test-group-id',
+              },
+              monitorReplicationFailures: true,
+              objectSizeMetrics: [100, 1000],
+            },
+            {},
+            { topic: 'metrics-test-topic' },
+            { topic: 'backbeat-gc' });
+        replicationStatusProcessor.start({ bootstrap: true }, () => {
+            assert(replicationStatusProcessor.getStateVars().gcProducer);
+            replicationStatusProcessor.stop(done);
+        });
+    });
+
+    it('should not create a GC if config is not provided', done => {
+        const replicationStatusProcessor = new ReplicationStatusProcessor(
+            { hosts: 'localhost:9092' },
+            { auth: { type: 'role',
+                      vault: { host: constants.source.vault,
+                               port: 7777 } },
+              s3: { host: constants.source.s3,
+                    port: 7777 },
+              transport: 'http',
+            },
+            { replicationStatusTopic:
+              'backbeat-func-test-repstatus',
+              replicationStatusProcessor: {
+                  retry: {
+                      timeoutS: 5,
+                  },
+                  groupId: 'backbeat-func-test-group-id',
+              },
+              monitorReplicationFailures: true,
+              objectSizeMetrics: [100, 1000],
+            },
+            {},
+            { topic: 'metrics-test-topic' });
+        replicationStatusProcessor.start({ bootstrap: true }, () => {
+            assert.strictEqual(replicationStatusProcessor.getStateVars().gcProducer, null);
+            replicationStatusProcessor.stop(done);
+        });
+    });
+});