Revert "Fix bug in processor"

This reverts commit 460fc4e.

Author: ThomasKranitsas

README.md

@@ -13,7 +13,6 @@ 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,7 +6,6 @@ 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.4.3",
+    "no-kafka": "^3.2.4",
     "topcoder-healthcheck-dropin": "^1.0.2",
     "winston": "^2.2.0"
   },

package.json

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