updated to match logstash changes

Author: joekiller

lib/logstash/inputs/kafka.rb

@@ -24,6 +24,12 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   # Specifies the ZooKeeper connection string in the form hostname:port where host and port are
   # the host and port of a ZooKeeper server. You can also specify multiple hosts in the form
   # hostname1:port1,hostname2:port2,hostname3:port3.
+  #
+  # The server may also have a ZooKeeper chroot path as part of it's ZooKeeper connection string
+  # which puts its data under some path in the global ZooKeeper namespace. If so the consumer
+  # should use the same chroot path in its connection string. For example to give a chroot path of
+  # /chroot/path you would give the connection string as
+  # hostname1:port1,hostname2:port2,hostname3:port3/chroot/path.
   config :zk_connect, :validate => :string, :default => 'localhost:2181'
   # A string that uniquely identifies the group of consumer processes to which this consumer
   # belongs. By setting the same group id multiple processes indicate that they are all part of
@@ -34,6 +40,11 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   # Specify whether to jump to beginning of the queue when there is no initial offset in
   # ZooKeeper, or if an offset is out of range. If this is false, messages are consumed
   # from the latest offset
+  #
+  # If reset_beginning is true, the consumer will check ZooKeeper to see if any other group members
+  # are present and active. If not, the consumer deletes any offset information in the ZooKeeper
+  # and starts at the smallest offset. If other group members are present reset_beginning will not
+  # work and the consumer threads will rejoin the consumer group.
   config :reset_beginning, :validate => :boolean, :default => false
   # Number of threads to read from the partitions. Ideally you should have as many threads as the
   # number of partitions for a perfect balance. More threads than partitions means that some
@@ -69,7 +80,7 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
 
   public
   def register
-    jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/kafka*/libs/*.jar")
+    jarpath = File.join(File.dirname(__FILE__), '../../../vendor/jar/kafka*/libs/*.jar')
     Dir[jarpath].each do |jar|
       require jar
     end
@@ -86,7 +97,7 @@ def register
         :consumer_id => @consumer_id,
         :fetch_message_max_bytes => @fetch_message_max_bytes
     }
-    if @reset_beginning == true
+    if @reset_beginning
       options[:reset_beginning] = 'from-beginning'
     end # if :reset_beginning
     @kafka_client_queue = SizedQueue.new(@queue_size)
@@ -107,7 +118,7 @@ def run(logstash_queue)
         end
       rescue LogStash::ShutdownSignal
         @logger.info('Kafka got shutdown signal')
-        @consumer_group.shutdown()
+        @consumer_group.shutdown
       end
       until @kafka_client_queue.empty?
         queue_event("#{@kafka_client_queue.pop}",logstash_queue)
@@ -117,7 +128,7 @@ def run(logstash_queue)
       @logger.warn('kafka client threw exception, restarting',
                    :exception => e)
       if @consumer_group.running?
-        @consumer_group.shutdown()
+        @consumer_group.shutdown
       end
       sleep(Float(@consumer_restart_sleep_ms) * 1 / 1000)
       retry
@@ -131,13 +142,13 @@ def queue_event(msg, output_queue)
       @codec.decode(msg) do |event|
         decorate(event)
         if @decorate_events
-          event['kafka'] = {'msg_size' => msg.bytesize, 'topic' => @topic_id, 'consumer_group' => @group_id}
+          event['kafka'] = {:msg_size => msg.bytesize, :topic => @topic_id, :consumer_group => @group_id}
         end
         output_queue << event
       end # @codec.decode
     rescue => e # parse or event creation error
-      @logger.error("Failed to create event", :message => msg, :exception => e,
-                    :backtrace => e.backtrace);
+      @logger.error('Failed to create event', :message => msg, :exception => e,
+                    :backtrace => e.backtrace)
     end # begin
   end # def queue_event
 

lib/logstash/outputs/kafka.rb

@@ -100,11 +100,11 @@ class LogStash::Outputs::Kafka < LogStash::Outputs::Base
   config :send_buffer_bytes, :validate => :number, :default => 100 * 1024
   # The client id is a user-specified string sent in each request to help trace calls. It should
   # logically identify the application making the request.
-  config :client_id, :validate => :string, :default => ""
+  config :client_id, :validate => :string, :default => ''
 
   public
   def register
-    jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/kafka*/libs/*.jar")
+    jarpath = File.join(File.dirname(__FILE__), '../../../vendor/jar/kafka*/libs/*.jar')
     Dir[jarpath].each do |jar|
       require jar
     end
@@ -130,13 +130,13 @@ def register
         :client_id => @client_id
     }
     @producer = Kafka::Producer.new(options)
-    @producer.connect()
+    @producer.connect
 
     @logger.info('Registering kafka producer', :topic_id => @topic_id, :broker_list => @broker_list)
 
     @codec.on_event do |event|
       begin
-        @producer.sendMsg(@topic_id,nil,event)
+        @producer.send_msg(@topic_id,nil,event)
       rescue LogStash::ShutdownSignal
         @logger.info('Kafka producer got shutdown signal')
       rescue => e

spec/inputs/kafka.rb

@@ -7,17 +7,20 @@
 require 'logstash/errors'
 
 describe LogStash::Inputs::Kafka do
+  extend LogStash::RSpec
+
+  let (:kafka_config) {{"topic_id" => "test"}}
 
   it 'should populate kafka config with default values' do
-    kafka = LogStash::Inputs::Kafka.new
+    kafka = LogStash::Inputs::Kafka.new(kafka_config)
     insist {kafka.zk_connect} == "localhost:2181"
     insist {kafka.topic_id} == "test"
     insist {kafka.group_id} == "logstash"
     insist {kafka.reset_beginning} == false
   end
 
-  it "should load register and load kafka jars without errors" do
-    kafka = LogStash::Inputs::Kafka.new
+  it "should register and load kafka jars without errors" do
+    kafka = LogStash::Inputs::Kafka.new(kafka_config)
     kafka.register
   end
 
@@ -33,7 +36,7 @@ def queue_event(msg, output_queue)
       end
     end
 
-    kafka = LogStash::Inputs::TestKafka.new
+    kafka = LogStash::Inputs::TestKafka.new(kafka_config)
     kafka.register
 
     class Kafka::Group
@@ -47,6 +50,8 @@ def run(a_numThreads, a_queue)
     kafka.run logstash_queue
     e = logstash_queue.pop
     insist { e["message"] } == "Kafka message"
-    insist { e["kafka"] } == {"msg_size"=>13, "topic"=>"test", "consumer_group"=>"logstash"}
+    # no metadata by default
+    insist { e["kafka"] } == nil
   end
+
 end

spec/outputs/kafka.rb

@@ -3,13 +3,15 @@
 require 'rspec'
 require 'insist'
 require 'logstash/namespace'
-require 'time'
+require "logstash/timestamp"
 require 'logstash/outputs/kafka'
 
 describe LogStash::Outputs::Kafka do
 
+  let (:kafka_config) {{"topic_id" => "test"}}
+
   it 'should populate kafka config with default values' do
-    kafka = LogStash::Outputs::Kafka.new
+    kafka = LogStash::Outputs::Kafka.new(kafka_config)
     insist {kafka.broker_list} == "localhost:9092"
     insist {kafka.topic_id} == "test"
     insist {kafka.compression_codec} == "none"
@@ -18,18 +20,18 @@
     insist {kafka.producer_type} == "sync"
   end
 
-  it "should load register and load kafka jars without errors" do
-    kafka = LogStash::Outputs::Kafka.new
+  it "should register and load kafka jars without errors" do
+    kafka = LogStash::Outputs::Kafka.new(kafka_config)
     kafka.register
   end
 
   it "should send logstash event to kafka broker" do
-    timestamp = Time.now
+    timestamp = LogStash::Timestamp.now
     expect_any_instance_of(Kafka::Producer)
-    .to receive(:sendMsg)
-        .with("test", nil, "{\"message\":\"hello world\",\"host\":\"test\",\"@timestamp\":\"#{timestamp.iso8601(3)}\",\"@version\":\"1\"}")
+    .to receive(:send_msg)
+        .with("test", nil, "{\"message\":\"hello world\",\"host\":\"test\",\"@timestamp\":\"#{timestamp}\",\"@version\":\"1\"}")
     e = LogStash::Event.new({"message" => "hello world", "host" => "test", "@timestamp" => timestamp})
-    kafka = LogStash::Outputs::Kafka.new
+    kafka = LogStash::Outputs::Kafka.new(kafka_config)
     kafka.register
     kafka.receive(e)
   end