Merge pull request #39 from topcoder-platform/Issue_38

Replace kafka simple consumer with group consumer

Author: Dushyant Bhalgami

README.md

@@ -20,6 +20,7 @@ The following parameters can be set in config files or in env variables:
 - KAFKA_CLIENT_CERT_KEY: Kafka connection private key, optional;
     if not provided, then SSL connection is not used, direct insecure connection is used;
     if provided, it can be either path to private key file or private key content
+- KAFKA_GROUP_ID: the Kafka group id, default value is 'submission-processor'
 - SUBMISSION_CREATE_TOPIC: Kafka topic related to Submission creation, default value is 'submission.notification.create'
 - AVSCAN_TOPIC: Kafka topic related to AV Scan, default value is 'avscan.action.scan'
 - ACCESS_KEY_ID: the AWS access key id

config/default.js

@@ -6,6 +6,9 @@ module.exports = {
   LOG_LEVEL: process.env.LOG_LEVEL || 'debug',
   MAXFILESIZE: process.env.MAXFILESIZE || 4294967296,
 
+  // Kafka group id
+  KAFKA_GROUP_ID: process.env.KAFKA_GROUP_ID || 'submission-processor',
+
   KAFKA_URL: process.env.KAFKA_URL || 'localhost:9092',
   KAFKA_CLIENT_CERT: process.env.KAFKA_CLIENT_CERT,
   KAFKA_CLIENT_CERT_KEY: process.env.KAFKA_CLIENT_CERT_KEY,

src/app.js

@@ -9,14 +9,13 @@ const Kafka = require('no-kafka')
 const co = require('co')
 const ProcessorService = require('./services/ProcessorService')
 const healthcheck = require('topcoder-healthcheck-dropin')
-const _ = require('lodash')
 
 // create consumer
-const options = { connectionString: config.KAFKA_URL }
+const options = { connectionString: config.KAFKA_URL, groupId: config.KAFKA_GROUP_ID }
 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) => {
@@ -47,14 +46,12 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, (
     return
   }
 
-
   if (topic === config.SUBMISSION_CREATE_TOPIC && messageJSON.payload.fileType === 'url') {
     logger.debug(`Ignoring message in topic ${messageJSON.topic} with file type as url`)
     // ignore the message
     return
   }
 
-
   return co(function * () {
     switch (topic) {
       case config.SUBMISSION_CREATE_TOPIC:
@@ -67,9 +64,14 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, (
         throw new Error(`Invalid topic: ${topic}`)
     }
   })
-    // commit offset
-    .then(() => consumer.commitOffset({ topic, partition, offset: m.offset }))
-    .catch((err) => logger.error(err))
+  // commit offset regardless of errors
+    .then(() => {
+      consumer.commitOffset({ topic, partition, offset: m.offset })
+    })
+    .catch((err) => {
+      logger.error(err)
+      consumer.commitOffset({ topic, partition, offset: m.offset })
+    })
 })
 
 // check if there is kafka connection alive
@@ -85,12 +87,18 @@ function check () {
   return connected
 }
 
+const topics = [config.SUBMISSION_CREATE_TOPIC, config.AVSCAN_TOPIC]
+
 consumer
-  .init()
-  // consume configured topic
+  .init([{
+    subscriptions: topics,
+    handler: dataHandler
+  }])
+  // consume configured topics
   .then(() => {
+    logger.info('Initialized.......')
     healthcheck.init([check])
-    const topics = [config.SUBMISSION_CREATE_TOPIC, config.AVSCAN_TOPIC]
-    _.each(topics, (tp) => consumer.subscribe(tp, { time: Kafka.LATEST_OFFSET }, dataHandler))
+    logger.info('Adding topics successfully.......')
+    logger.info(topics)
+    logger.info('Kick Start.......')
   })
-  .catch((err) => logger.error(err))