add producer for try plugin

Author: Filippo Balicchia

bin/producer.js

@@ -0,0 +1,25 @@
+var kafka_native = require('kafka-native');
+
+var broker = 'localhost:9092';
+var topic = 'test';
+
+function produceMessage() {
+    var producer = new kafka_native.Producer({
+        broker: broker
+    });
+
+    producer.partition_count(topic)
+    .then(function(npartitions) {
+        var partition = 0;
+        setInterval(function() {
+            for(var i = 0; i < 100; ++i){
+                message = 'messageProduced-' + i 
+                console.log('producer send message ' + message + ' to partition' + partition); 
+                producer.send(topic, partition, [message]);
+                partition = (partition + 1) % npartitions;
+            }
+        }, 1000);
+    });
+}
+
+produceMessage();

config/examples/kafka-stdout-yml.yml

@@ -7,7 +7,7 @@ input:
     brokerAddress: localhost
     offset_directory: /tmp/kafka-offsets
     brokerPort: 9092
-    topic: test4
+    topic: test
     debug: true
 
 output:

docker-compose.yml

@@ -0,0 +1,16 @@
+zookeeper:
+  image: wurstmeister/zookeeper
+  ports:
+    - 2181:2181
+kafka:
+  image: wurstmeister/kafka:0.9.0.1
+  ports:
+    - 9092:9092
+  links:
+    - zookeeper
+  environment:
+    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+    KAFKA_CREATE_TOPICS: "test:1:1"
+    KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
+    KAFKA_ADVERTISED_PORT: 9092
+    KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'

lib/plugins/input/kafka.js

@@ -11,7 +11,7 @@ var consoleLogger = require('../../util/logger.js')
 function InputKafka (config, eventEmitter) {
   this.config = config
   this.eventEmitter = eventEmitter
-  this.consumer;
+  ;
 }
 module.exports = InputKafka
 /**
@@ -29,14 +29,11 @@ InputKafka.prototype.start = function () {
 
 
 InputKafka.prototype.createServer = function () {
-
-    consoleLogger.log('start kafka consumer ')
-    
+    consoleLogger.log('start kafkaConsumer ')
     var brokerVar = this.config.brokerAddress + ':' + this.config.brokerPort
     var topicVar = this.config.topic;
     var eventEmitter = this.eventEmitter
-    consoleLogger.log('Init kafka consumer ')
-    consumer = new kafka.Consumer({
+    global.kafkaConsumer = new kafka.Consumer({
     broker: brokerVar,
     topic: topicVar,
     offset_directory: this.config.offset_directory,
@@ -48,8 +45,7 @@ InputKafka.prototype.createServer = function () {
     }
     });
 
-    consoleLogger.log('start consumer ')
-    consumer.start()
+    kafkaConsumer.start()
 
 }
 
@@ -58,8 +54,8 @@ InputKafka.prototype.createServer = function () {
  * we close kafka consumer.
  */
 InputKafka.prototype.stop = function (cb) {
-    consoleLogger.log('kafka input stop')
-    consumer.stop()
+    consoleLogger.log('stop kafkaConsumer')
+    kafkaConsumer.stop()
     cb()
 }