Fix bug in processor

Author: ThomasKranitsas

README.md

@@ -13,6 +13,7 @@ Configuration for the notification server is at `config/default.js`.
 The following parameters can be set in config files or in env variables:
 
 - LOG_LEVEL: the log level; default value: 'debug'
+- KAFKA_GROUP_ID: group id of the consumer; default value: 'submission-processor-es-group'
 - KAFKA_URL: comma separated Kafka hosts; default value: 'localhost:9092'
 - KAFKA_CLIENT_CERT: Kafka connection certificate, optional; default value is undefined;
     if not provided, then SSL connection is not used, direct insecure connection is used;

config/default.js

@@ -6,6 +6,7 @@ module.exports = {
   DISABLE_LOGGING: process.env.DISABLE_LOGGING || false, // If true, logging will be disabled
   LOG_LEVEL: process.env.LOG_LEVEL || 'debug',
 
+  KAFKA_GROUP_ID: process.env.KAFKA_GROUP_ID || 'submission-processor-es-group',
   KAFKA_URL: process.env.KAFKA_URL || 'localhost:9092',
   // below are used for secure Kafka connection, they are optional
   // for the local Kafka, they are not needed

package-lock.json

@@ -35,7 +35,7 @@
     "http-aws-es": "^6.0.0",
     "joi": "^9.0.4",
     "lodash": "^4.17.10",
-    "no-kafka": "^3.2.4",
+    "no-kafka": "^3.4.3",
     "topcoder-healthcheck-dropin": "^1.0.2",
     "winston": "^2.2.0"
   },

package.json

@@ -3,23 +3,21 @@
  */
 
 global.Promise = require('bluebird')
-const _ = require('lodash')
 const config = require('config')
 const logger = require('./common/logger')
 const Kafka = require('no-kafka')
-const co = require('co')
 const ProcessorService = require('./services/ProcessorService')
 const healthcheck = require('topcoder-healthcheck-dropin')
 
 // create consumer
-const options = { connectionString: config.KAFKA_URL, handlerConcurrency: 1 }
+const options = { connectionString: config.KAFKA_URL, groupId: config.KAFKA_GROUP_ID, handlerConcurrency: 1 }
 if (config.KAFKA_CLIENT_CERT && config.KAFKA_CLIENT_CERT_KEY) {
   options.ssl = { cert: config.KAFKA_CLIENT_CERT, key: config.KAFKA_CLIENT_CERT_KEY }
 }
-const consumer = new Kafka.SimpleConsumer(options)
+const consumer = new Kafka.GroupConsumer(options)
 
 // data handler
-const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, (m) => {
+const dataHandler = async (messageSet, topic, partition) => Promise.each(messageSet, async (m) => {
   const message = m.message.value.toString('utf8')
   logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${
     m.offset}; Message: ${message}.`)
@@ -37,24 +35,26 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, (
     // ignore the message
     return
   }
-  return co(function * () {
+  try {
     switch (topic) {
       case config.CREATE_DATA_TOPIC:
-        yield ProcessorService.create(messageJSON)
+        await ProcessorService.create(messageJSON)
         break
       case config.UPDATE_DATA_TOPIC:
-        yield ProcessorService.update(messageJSON)
+        await ProcessorService.update(messageJSON)
         break
       case config.DELETE_DATA_TOPIC:
-        yield ProcessorService.remove(messageJSON)
+        await ProcessorService.remove(messageJSON)
         break
       default:
         throw new Error(`Invalid topic: ${topic}`)
     }
-  })
+
     // commit offset
-    .then(() => consumer.commitOffset({ topic, partition, offset: m.offset }))
-    .catch((err) => logger.error(err))
+    await consumer.commitOffset({ topic, partition, offset: m.offset })
+  } catch (err) {
+    logger.error(err)
+  }
 })
 
 // check if there is kafka connection alive
@@ -70,13 +70,14 @@ function check () {
   return connected
 }
 
+const topics = [config.CREATE_DATA_TOPIC, config.UPDATE_DATA_TOPIC, config.DELETE_DATA_TOPIC]
+// consume configured topics
 consumer
-  .init()
-  // consume configured topics
+  .init([{
+    subscriptions: topics,
+    handler: dataHandler
+  }])
   .then(() => {
     healthcheck.init([check])
-
-    const topics = [config.CREATE_DATA_TOPIC, config.UPDATE_DATA_TOPIC, config.DELETE_DATA_TOPIC]
-    _.each(topics, (tp) => consumer.subscribe(tp, { time: Kafka.LATEST_OFFSET }, dataHandler))
   })
   .catch((err) => logger.error(err))