Merge pull request #6 from topcoder-platform/debug-es

Debug why ES will not update

Author: cwdcwd

package-lock.json

@@ -27,6 +27,7 @@
   },
   "dependencies": {
     "@hapi/joi": "^15.1.0",
+    "async-mutex": "^0.2.4",
     "aws-sdk": "^2.476.0",
     "bluebird": "^3.5.5",
     "config": "^3.1.0",

package.json

@@ -9,12 +9,28 @@ const healthcheck = require('topcoder-healthcheck-dropin')
 const logger = require('./common/logger')
 const helper = require('./common/helper')
 const ProcessorService = require('./services/ProcessorService')
+const Mutex = require('async-mutex').Mutex
 
 // Start kafka consumer
 logger.info('Starting kafka consumer')
 // create consumer
 const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
 
+let count = 0
+let mutex = new Mutex()
+
+async function getLatestCount () {
+  const release = await mutex.acquire()
+
+  try {
+    count = count + 1
+
+    return count
+  } finally {
+    release()
+  }
+}
+
 /*
  * Data handler linked with Kafka consumer
  * Whenever a new message is received by Kafka consumer,
@@ -25,12 +41,17 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${
     m.offset}; Message: ${message}.`)
   let messageJSON
+  let messageCount = await getLatestCount()
+
+  logger.debug(`Current message count: ${messageCount}`)
   try {
     messageJSON = JSON.parse(message)
   } catch (e) {
     logger.error('Invalid message JSON.')
     logger.logFullError(e)
 
+    logger.debug(`Commiting offset after processing message with count ${messageCount}`)
+
     // commit the message and ignore it
     await consumer.commitOffset({ topic, partition, offset: m.offset })
     return
@@ -39,6 +60,8 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   if (messageJSON.topic !== topic) {
     logger.error(`The message topic ${messageJSON.topic} doesn't match the Kafka topic ${topic}.`)
 
+    logger.debug(`Commiting offset after processing message with count ${messageCount}`)
+
     // commit the message and ignore it
     await consumer.commitOffset({ topic, partition, offset: m.offset })
     return
@@ -57,10 +80,12 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
         break
     }
 
-    logger.debug('Successfully processed message')
+    logger.debug(`Successfully processed message with count ${messageCount}`)
   } catch (err) {
     logger.logFullError(err)
   } finally {
+    logger.debug(`Commiting offset after processing message with count ${messageCount}`)
+
     // Commit offset regardless of error
     await consumer.commitOffset({ topic, partition, offset: m.offset })
   }