Merge pull request #25 from afrisalyp/issue-221

Change simple consumer to group consumer.

Author: jmgasper

README.md

@@ -57,6 +57,7 @@ For using with SSL, the options should be as
 ```
  {
     connectionString: '<server>',
+    groupId: <groupid>,
     ssl: {
       cert: '<certificate>',
       key:  '<key>'

config/default.js

@@ -18,6 +18,7 @@ module.exports = {
   TOPIC: process.env.TOPIC || 'tc-x-events',
   KAFKA_OPTIONS: {
     connectionString: process.env.KAFKA_URL || 'localhost:9092',
+    groupId: process.env.KAFKA_GROUP_ID || 'topcoder-x-processor',
     ssl: {
       cert: process.env.KAFKA_CLIENT_CERT || fs.readFileSync('./kafka_client.cer'), // eslint-disable-line no-sync
       key: process.env.KAFKA_CLIENT_CERT_KEY || fs.readFileSync('./kafka_client.key'), // eslint-disable-line no-sync

configuration.md

@@ -9,6 +9,7 @@ The following config parameters are supported, they are defined in `config/defau
 |TOPIC  | The Kafka topic where events are published.  This must be the same as the configured value for topcoder-x-processor| |
 |KAFKA_OPTIONS | Kafka connection options| |
 |KAFKA_URL | The Kafka host to connect to| localhost:9092 |
+|KAFKA_GROUP_ID | The Kafka group id name| topcoder-x-processor |
 |KAFKA_CLIENT_CERT | The Kafka SSL certificate to use when connecting| Read from kafka_client.cer file, but this can be set as a string like it is on Heroku |
 |KAFKA_CLIENT_CERT_KEY | The Kafka SSL certificate key to use when connecting| Read from kafka_client.key file, but this can be set as a string like it is on Heroku|
 |TC_DEV_ENV| the flag whether to use topcoder development api or production| false|

utils/kafka.js

@@ -21,7 +21,7 @@ const logger = require('./logger');
 
 class Kafka {
   constructor() {
-    this.consumer = new kafka.SimpleConsumer(config.KAFKA_OPTIONS);
+    this.consumer = new kafka.GroupConsumer(config.KAFKA_OPTIONS);
 
     this.producer = new kafka.Producer(config.KAFKA_OPTIONS);
     this.producer.init().then(() => {
@@ -77,10 +77,12 @@ class Kafka {
   }
 
   run() {
-    this.consumer.init().then(() => {
+    this.consumer.init([{
+      subscriptions: [config.TOPIC],
+      handler: this.messageHandler
+    }]).then(() => {
       logger.info('kafka consumer is ready');
       healthcheck.init([this.check]);
-      this.consumer.subscribe(config.TOPIC, {}, this.messageHandler);
     }).catch((err) => {
       logger.error(`kafka consumer is not connected. ${err.stack}`);
     });