Check if GC config exists before creating gc producer in replication

extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js

@@ -204,14 +204,16 @@
      *   in PEM format
      * @param {Object} mConfig - metrics config
      * @param {String} mConfig.topic - metrics config kafka topic
+     * @param {Boolean} gcConfig - flag to enable garbage collection - Based on GC config
      */
     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 @@
                 });
             },
             done => {
-                this._gcProducer = new GarbageCollectorProducer();
-                this._gcProducer.setupProducer(done);
+                if (this.gcConfig) {
+                    this._gcProducer = new GarbageCollectorProducer();
+                    return this._gcProducer.setupProducer(done);
+                } else {
+                    return 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

@@ -843,7 +843,7 @@ describe('queue processor functional tests with mocking', () => {
                           objectSizeMetrics: [100, 1000],
                         },
                         {},
-                        { topic: 'metrics-test-topic' });
+                        { topic: 'metrics-test-topic' }, { topic: 'backbeat-gc' });
                     replicationStatusProcessor.start({ bootstrap: true }, done);
                 },
                 done => {

tests/unit/replication/ReplicationStatusProcessor.js

@@ -12,7 +12,7 @@ function makeReplicationStatusProcessor(replayTopics) {
             transport: 'http',
         },
         { replayTopics, objectSizeMetrics: [100, 1000] },
-        {});
+        {}, false);
 }
 
 describe('ReplicationStatusProcessor', () => {