use retry value from payload

Author: eisbilir

src/common/helper.js

@@ -12,7 +12,6 @@ const _ = require('lodash')
 const { Mutex } = require('async-mutex')
 const m2mAuth = require('tc-core-library-js').auth.m2m
 const busApi = require('@topcoder-platform/topcoder-bus-api-wrapper')
-const ActionProcessorService = require('../services/ActionProcessorService')
 
 AWS.config.region = config.esConfig.AWS_REGION
 
@@ -185,10 +184,11 @@ async function postMessageViaWebhook (webhook, message) {
  * Calls ActionProcessorService to attempt to retry failed process
  * @param {String} topic the failed topic name
  * @param {Object} payload the payload
- * @param {String} id the id that was the subject of the operation failed
+ * @param {String} retry how many times has it been retried
  */
-async function retryFailedProcess (topic, payload, id) {
-  await ActionProcessorService.processCreate(topic, payload, id)
+async function retryFailedProcess (topic, payload, retry) {
+  const ActionProcessorService = require('../services/ActionProcessorService')
+  await ActionProcessorService.processCreate(topic, payload, retry)
 }
 
 let busApiClient

src/services/ActionProcessorService.js

@@ -6,14 +6,11 @@ const Joi = require('@hapi/joi')
 const logger = require('../common/logger')
 const helper = require('../common/helper')
 const config = require('config')
-const _ = require('lodash')
 
 const localLogger = {
   debug: ({ context, message }) => logger.debug({ component: 'ActionProcessorService', context, message })
 }
 
-const retryMap = {}
-
 /**
   * Process retry operation message
   * @param {Object} message the kafka message
@@ -25,9 +22,10 @@ async function processRetry (message, transactionId) {
     return
   }
   const { topicServiceMapping } = require('../app')
+  const retry = message.payload.retry
   message.topic = message.payload.originalTopic
   message.payload = message.payload.originalPayload
-  await topicServiceMapping[message.topic](message, transactionId)
+  await topicServiceMapping[message.topic](message, transactionId, { retry })
 }
 
 processRetry.schema = {
@@ -49,30 +47,29 @@ processRetry.schema = {
   * Analyzes the failed process and sends it to bus api to be received again.
   * @param {String} originalTopic the failed topic name
   * @param {Object} originalPayload the payload
-  * @param {String} id the id that was the subject of the operation failed
+  * @param {Number} retry how many times has it been retried
   */
-async function processCreate (originalTopic, originalPayload, id) {
-  const retry = _.defaultTo(retryMap[id], 0) + 1
+async function processCreate (originalTopic, originalPayload, retry) {
+  retry = retry + 1
   if (retry > config.MAX_RETRY) {
-    localLogger.debug({ context: 'processCreate', message: `retry: ${retry} for ${id} exceeds the max retry: ${config.MAX_RETRY} - ignored` })
+    localLogger.debug({ context: 'processCreate', message: `retry: ${retry} for ${originalPayload.id} exceeds the max retry: ${config.MAX_RETRY} - ignored` })
     return
   }
-  localLogger.debug({ context: 'processCreate', message: `retry: ${retry} for ${id}` })
-  retryMap[id] = retry
+  localLogger.debug({ context: 'processCreate', message: `retry: ${retry} for ${originalPayload.id}` })
   const payload = {
     originalTopic,
     originalPayload,
     retry
   }
-  setTimeout(async function () {
+  setTimeout(async () => {
     await helper.postEvent(config.topics.TAAS_ACTION_RETRY_TOPIC, payload)
   }, 2 ** retry * config.BASE_RETRY_DELAY)
 }
 
 processCreate.schema = {
   originalTopic: Joi.string().required(),
   originalPayload: Joi.object().required(),
-  id: Joi.string().uuid().required()
+  retry: Joi.number().integer().min(0).required()
 }
 
 module.exports = {

src/services/WorkPeriodProcessorService.js

@@ -14,8 +14,9 @@ const esClient = helper.getESClient()
  * Process create entity message
  * @param {Object} message the kafka message
  * @param {String} transactionId
+ * @param {Object} options
  */
-async function processCreate (message, transactionId) {
+async function processCreate (message, transactionId, options) {
   const workPeriod = message.payload
   // Find related resourceBooking
   let resourceBooking
@@ -30,7 +31,7 @@ async function processCreate (message, transactionId) {
     // it has not yet been created. We should send a retry request.
     if (err.httpStatus === 404) {
       logger.logFullError(err, { component: 'WorkPeriodProcessorService', context: 'processCreate' })
-      await helper.retryFailedProcess(message.topic, workPeriod, workPeriod.resourceBookingId)
+      await helper.retryFailedProcess(message.topic, workPeriod, options.retry)
       return
     } else {
       throw err
@@ -78,7 +79,12 @@ processCreate.schema = {
       updatedBy: Joi.string().uuid().allow(null)
     }).required()
   }).required(),
-  transactionId: Joi.string().required()
+  transactionId: Joi.string().required(),
+  options: Joi.object().keys({
+    retry: Joi.number().integer().min(0).default(0)
+  }).default({
+    retry: 0
+  })
 }
 
 /**