Kafka node sample program

var kafka = require('kafka-node'),
    express = require('express'),
    app = express(),
    bp = require('body-parser'),
    Producer = kafka.Producer,
    Consumer = kafka.Consumer,
    KeyedMessage = kafka.KeyedMessage,
    client = new kafka.Client(),
    producer = new Producer(client);
    var topic="test";


producer.on('error', function (err) {
  //console.log(err);
})

producer.on('ready', function () {
    readMessages();
});




function readMessages(){
try{
 var consumer = new Consumer(
        client,[{ topic:topic, partition: 1 }],{autoCommit: true}
 );
 consumer.on('message', function (message) {
    try{
        console.log(message);
    }
    catch(ex){
       console.log("error",ex);
    }
  });
}
catch(ex){
  console.log(ex);
}

}


process.on('uncaughtException', function (err) {
  console.log('Caught exception: ' + err);
});