Fixed and working. Still issue with race-condition/sync. Re-enabling retry for now.

Author: rootelement

src/app.js

@@ -8,7 +8,7 @@ const { isNil, get } = require('lodash')
 const Kafka = require('no-kafka')
 const healthcheck = require('topcoder-healthcheck-dropin')
 const logger = require('./common/logger')
-// const helper = require('./common/helper')
+const helper = require('./common/helper')
 const { getKafkaOptions } = require('./common/utils')
 const ProcessorService = require('./services/ProcessorService')
 
@@ -58,26 +58,25 @@ const dataHandler = async (messageSet, topic, partition) => Promise.each(message
       if (isNil(challengeId)) {
         throw new Error(`Challenge ID ${challengeId} is null, will not queue to retry`)
       } else {
-        logger.info('Should Retry - but Retry is Disabled')
-        // const retryCountIdentifier = `${config.KAFKA_GROUP_ID.split(' ').join('_')}_retry_count`
-        // let currentRetryCount = parseInt(get(messageJSON.payload, retryCountIdentifier, 1), 10)
-        // if (currentRetryCount <= config.MAX_RETRIES) {
-        //   logger.info(`Challenge does not exist yet. Will post the same message back to the bus API and retry in ${currentRetryCount * (config.RETRY_TIMEOUT / 1000)} seconds`)
-        //   await new Promise((resolve) => {
-        //     setTimeout(async () => {
-        //       currentRetryCount += 1
-        //       await helper.postBusEvent(topic, { ...messageJSON.payload, [retryCountIdentifier]: currentRetryCount })
-        //       resolve()
-        //     }, config.RETRY_TIMEOUT * currentRetryCount)
-        //   })
-        // } else {
-        //   logger.error(`Failed to process message after ${config.MAX_RETRIES} retries. Aborting...`)
-        // }
+        const retryCountIdentifier = `${config.KAFKA_GROUP_ID.split(' ').join('_')}_retry_count`
+        let currentRetryCount = parseInt(get(messageJSON.payload, retryCountIdentifier, 1), 10)
+        if (currentRetryCount <= config.MAX_RETRIES) {
+          logger.info(`Challenge does not exist yet. Will post the same message back to the bus API and retry in ${currentRetryCount * (config.RETRY_TIMEOUT / 1000)} seconds`)
+          await new Promise((resolve) => {
+            setTimeout(async () => {
+              currentRetryCount += 1
+              await helper.postBusEvent(topic, { ...messageJSON.payload, [retryCountIdentifier]: currentRetryCount })
+              resolve()
+            }, config.RETRY_TIMEOUT * currentRetryCount)
+          })
+        } else {
+          logger.error(`Failed to process message after ${config.MAX_RETRIES} retries. Aborting...`)
+        }
       }
     }
     // only commit if no errors
     await consumer.commitOffset({ topic, partition, offset: m.offset })
-    logger.debug('Successfully processed message')
+    // logger.debug('Successfully processed message')
   } catch (err) {
     logger.error(`app.js error message: ${err.message}`)
   }

src/dao/RegistrationDAO.js

@@ -1,6 +1,6 @@
 const helper = require('../common/helper')
 const moment = require('moment')
-const logger = require('../common/logger')
+// const logger = require('../common/logger')
 
 const RESOURCE_TYPE_EXT_REF_ID = 1
 const RESOURCE_TYPE_HANDLE_ID = 2
@@ -41,7 +41,7 @@ VALUES
 
 async function persistResourceWithRoleId (userId, challengeId, resourceId, roleId, handle, projectPhaseId) {
   const regDate = moment().format('MM[.]DD[.]YYYY h:mm A')
-  logger.debug(`persistResourceWithRoleId - projectPhaseId: ${projectPhaseId} - ${JSON.stringify([resourceId, roleId, projectPhaseId, challengeId, userId, userId, userId])}`)
+  // logger.debug(`persistResourceWithRoleId - projectPhaseId: ${projectPhaseId} - ${JSON.stringify([resourceId, roleId, projectPhaseId, challengeId, userId, userId, userId])}`)
   await helper.executeSQLonDB(QUERY_INSERT_RESOURCE_WITH_ROLE, [resourceId, roleId, projectPhaseId, challengeId, userId, userId, userId])
 
   await persistResourceInfo(userId, resourceId, RESOURCE_TYPE_EXT_REF_ID, userId)

src/services/ProcessorService.js

@@ -24,8 +24,8 @@ async function legacyChallengeExist (message) {
     const m2mToken = await helper.getM2Mtoken()
     const res = await helper.getRequest(`${config.CHALLENGE_API_V5_URL}/${challengeId}`, m2mToken)
     // logger.debug(`m2m Token: ${m2mToken}`)
-    logger.debug(`Getting Challenge from V5 ${config.CHALLENGE_API_V5_URL}/${challengeId}`)
-    logger.debug(`Response ${JSON.stringify(res.body)}`)
+    // logger.debug(`Getting Challenge from V5 ${config.CHALLENGE_API_V5_URL}/${challengeId}`)
+    // logger.debug(`Response ${JSON.stringify(res.body)}`)
     const v5Challenge = res.body
     if (!v5Challenge.legacyId) {
       exists = false
@@ -112,7 +112,7 @@ async function _updateChallengeResource (message, isDelete) {
       }
     } else {
       if (isDelete) {
-        logger.debug(`v4 Deleteing Challenge Resource ${config.CHALLENGE_API_V4_URL}/${_.get(v5Challenge, 'legacyId')}/resources - ${JSON.stringify(body)}`)
+        logger.debug(`v4 Deleting Challenge Resource ${config.CHALLENGE_API_V4_URL}/${_.get(v5Challenge, 'legacyId')}/resources - ${JSON.stringify(body)}`)
         response = await helper.deleteRequest(`${config.CHALLENGE_API_V4_URL}/${_.get(v5Challenge, 'legacyId')}/resources`, body, m2mToken)
       } else {
         logger.debug(`v4 Creating Challenge Resource ${config.CHALLENGE_API_V4_URL}/${_.get(v5Challenge, 'legacyId')}/resources - ${JSON.stringify(body)}`)
@@ -132,11 +132,11 @@ async function createChallengeResource (message) {
   try {
     await _updateChallengeResource(message, false)
   } catch (e) {
+    logger.error(e.message)
     logger.logFullError(e)
-    logger.debug(e.message)
   }
 
-  logger.info(`Successfully processed create challenge resource message : ${JSON.stringify(message)}`)
+  // logger.debug(`Successfully processed create challenge resource message : ${JSON.stringify(message)}`)
 }
 
 createChallengeResource.schema = {
@@ -162,10 +162,10 @@ async function deleteChallengeResource (message) {
   try {
     await _updateChallengeResource(message, true)
   } catch (e) {
-    logger.info(`Failed to find and delete the resource: ${JSON.stringify(message)} Error: ${JSON.stringify(e)}`)
+    logger.error(`Failed to find and delete the resource: ${JSON.stringify(message)} Error: ${JSON.stringify(e)}`)
   }
 
-  logger.info(`Successfully processed delete challenge resource message : ${JSON.stringify(message)}`)
+  // logger.debug(`Successfully processed delete challenge resource message : ${JSON.stringify(message)}`)
 }
 
 deleteChallengeResource.schema = createChallengeResource.schema

src/services/ProjectService.js

@@ -14,7 +14,7 @@ const QUERY_CHECK_RESOURCE_EXISTS = 'SELECT COUNT(*) as num FROM resource WHERE
  */
 async function resourceExists (challengeId, roleId, userId) {
   const result = helper.queryDataFromDB(QUERY_CHECK_RESOURCE_EXISTS, [challengeId, roleId, toInteger(userId)])
-  logger.debug(`resourceExists ${JSON.stringify([challengeId, roleId, toInteger(userId)])} result: ${JSON.stringify(result)}`)
+  // logger.debug(`resourceExists ${JSON.stringify([challengeId, roleId, toInteger(userId)])} result: ${JSON.stringify(result)}`)
   if (result && result.length > 0) {
     return result[0].num > 0
   }