Merge pull request #285 from scality/ft/ZENKO-147/configurableFailedC…

…RRTracking ft: ZENKO-147 Make CRR retry configurable
scality · May 25, 2018 · 92eb9ac · 92eb9ac
2 parents fbf8b2f + 45c9d41
commit 92eb9ac
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/extensions/replication/ReplicationConfigValidator.js b/extensions/replication/ReplicationConfigValidator.js
@@ -50,6 +50,7 @@ const joiSchema = {
     },
     topic: joi.string().required(),
     replicationStatusTopic: joi.string().required(),
+    monitorReplicationFailures: joi.boolean().default(true),
     queueProcessor: {
         groupId: joi.string().required(),
         retryTimeoutS: joi.number().default(300),

diff --git a/extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js b/extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js
@@ -182,7 +182,7 @@ class ReplicationStatusProcessor {
         if (sourceEntry instanceof ObjectQueueEntry) {
             task = new UpdateReplicationStatus(this);
         }
-        if (task) {
+        if (task && this.repConfig.monitorReplicationFailures) {
             return this._setFailedKeys(sourceEntry, kafkaEntry, err => {
                 if (err) {
                     this.logger.error('error setting redis hash key', {
@@ -193,6 +193,10 @@ class ReplicationStatusProcessor {
                     sourceEntry.getCanonicalKey(), done);
             });
         }
+        if (task) {
+            return this.taskScheduler.push({ task, entry: sourceEntry },
+                sourceEntry.getCanonicalKey(), done);
+        }
         this.logger.warn('skipping unknown source entry',
                          { entry: sourceEntry.getLogInfo() });
         return process.nextTick(done);