<NME> kafka.rb
<BEF> #require 'logstash/namespace'
#require 'logstash/outputs/base'
#require 'jruby-kafka'
#
#class LogStash::Outputs::Kafka < LogStash::Outputs::Base
#  config_name 'kafka'
#  milestone 1
#
#  default :codec, 'plain'
#
#  config :zk_connect, :validate => :string, :required => true
#  config :broker_list, :validate => :string, :required => true
#  config :topic_id, :validate => :string, :required => true
#
#  public
#  def register
#    jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/kafka*.jar")
#    Dir[jarpath].each do |jar|
#      require jar
#    end
#    options = {
#        :zk_connect => @zk_connect,
#        :topic_id => @topic_id,
#        :broker_list => @broker_list
#    }
#    @producer = Kafka::Producer.new(options)
#    @producer.connect()
#    @logger.info('Registering kafka', :topic_id => @topic_id, :zk_connect => @zk_connect)
#  end # def register
#
#  public
#  def receive(event)
#    return unless output?(event)
#
#    key = event.sprintf(@key)
#    # TODO(sissel): We really should not drop an event, but historically
#    # we have dropped events that fail to be converted to json.
#    # TODO(sissel): Find a way to continue passing events through even
#    # if they fail to convert properly.
#    begin
#      payload = event.to_json
#    rescue Encoding::UndefinedConversionError, ArgumentError
#      puts "FAILUREENCODING"
#      @logger.error("Failed to convert event to JSON. Invalid UTF-8, maybe?",
#                    :event => event.inspect)
#      return
#    end
#
#    @producer.sendMsg(key,payload)
#  end # def receive
#
#  public
#  def teardown
#    @kafka_client_queue.push(:stop_plugin)
#  end
#
#  private
#  def queue_event(msg, output_queue)
#    begin
#      @codec.decode(msg) do |event|
#        decorate(event)
#        event['kafka'] = {'msg_size' => msg.bytesize, 'topic' => @topic_id, 'consumer_group' => @group_id}
#        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);
#    end # begin
#  end # def queue_event
#
#end #class LogStash::Inputs::Kafka
  config :message_send_max_retries, :validate => :number, :default => 3
  # Before each retry, the producer refreshes the metadata of relevant topics to see if a new
  # leader has been elected. Since leader election takes a bit of time,
  # this property specifies the amount of time that the producer waits before refreshing the
  # metadata.
  config :retry_backoff_ms, :validate => :number, :default => 100
  # The producer generally refreshes the topic metadata from brokers when there is a failure
  # (partition missing, leader not available...). It will also poll regularly (default: every
  # 10min so 600000ms). If you set this to a negative value, metadata will only get refreshed on
  # failure. If you set this to zero, the metadata will get refreshed after each message sent
  # (not recommended). Important note: the refresh happen only AFTER the message is sent,
  # so if the producer never sends a message the metadata is never refreshed
  config :topic_metadata_refresh_interval_ms, :validate => :number, :default => 600 * 1000
  # Maximum time to buffer data when using async mode. For example a setting of 100 will try to
  # batch together 100ms of messages to send at once. This will improve throughput but adds
  # message delivery latency due to the buffering.
  config :queue_buffering_max_ms, :validate => :number, :default => 5000
  # The maximum number of unsent messages that can be queued up the producer when using async
  # mode before either the producer must be blocked or data must be dropped.
  config :queue_buffering_max_messages, :validate => :number, :default => 10000
  # The amount of time to block before dropping messages when running in async mode and the
  # buffer has reached queue.buffering.max.messages. If set to 0 events will be enqueued
  # immediately or dropped if the queue is full (the producer send call will never block). If set
  # to -1 the producer will block indefinitely and never willingly drop a send.
  config :queue_enqueue_timeout_ms, :validate => :number, :default => -1
  # The number of messages to send in one batch when using async mode. The producer will wait
  # until either this number of messages are ready to send or queue.buffer.max.ms is reached.
  config :batch_num_messages, :validate => :number, :default => 200
  # Socket write buffer size
  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 => ""

  public
  def register
    LogStash::Logger.setup_log4j(@logger)
    options = {
      :broker_list => @broker_list,
      :compression_codec => @compression_codec,
      :compressed_topics => @compressed_topics,
      :request_required_acks => @request_required_acks,
      :serializer_class => @serializer_class,
      :partitioner_class => @partitioner_class,
      :request_timeout_ms => @request_timeout_ms,
      :producer_type => @producer_type,
      :key_serializer_class => @key_serializer_class,
      :message_send_max_retries => @message_send_max_retries,
      :retry_backoff_ms => @retry_backoff_ms,
      :topic_metadata_refresh_interval_ms => @topic_metadata_refresh_interval_ms,
      :queue_buffering_max_ms => @queue_buffering_max_ms,
      :queue_buffering_max_messages => @queue_buffering_max_messages,
      :queue_enqueue_timeout_ms => @queue_enqueue_timeout_ms,
      :batch_num_messages => @batch_num_messages,
      :send_buffer_bytes => @send_buffer_bytes,
      :client_id => @client_id
    }
    @producer = Kafka::Producer.new(options)
    @producer.connect

    @logger.info('Registering kafka producer', :topic_id => @topic_id, :broker_list => @broker_list)

    @codec.on_event do |event|
      begin
        @producer.send_msg(@topic_id,nil,event)
      rescue LogStash::ShutdownSignal
        @logger.info('Kafka producer got shutdown signal')
      rescue => e
        @logger.warn('kafka producer threw exception, restarting',
                     :exception => e)
      end
    end
  end # def register

  def receive(event)
    return unless output?(event)
    if event == LogStash::SHUTDOWN
      finished
      return
    end
    @codec.encode(event)
  end
    
  def teardown
    @producer.close
  end

end #class LogStash::Outputs::Kafka

<MSG> adding producer back

<DFF> @@ -1,71 +1,71 @@
-#require 'logstash/namespace'
-#require 'logstash/outputs/base'
-#require 'jruby-kafka'
-#
-#class LogStash::Outputs::Kafka < LogStash::Outputs::Base
-#  config_name 'kafka'
-#  milestone 1
-#
-#  default :codec, 'plain'
-#
-#  config :zk_connect, :validate => :string, :required => true
-#  config :broker_list, :validate => :string, :required => true
-#  config :topic_id, :validate => :string, :required => true
-#
-#  public
-#  def register
-#    jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/kafka*.jar")
-#    Dir[jarpath].each do |jar|
-#      require jar
-#    end
-#    options = {
-#        :zk_connect => @zk_connect,
-#        :topic_id => @topic_id,
-#        :broker_list => @broker_list
-#    }
-#    @producer = Kafka::Producer.new(options)
-#    @producer.connect()
-#    @logger.info('Registering kafka', :topic_id => @topic_id, :zk_connect => @zk_connect)
-#  end # def register
-#
-#  public
-#  def receive(event)
-#    return unless output?(event)
-#
-#    key = event.sprintf(@key)
-#    # TODO(sissel): We really should not drop an event, but historically
-#    # we have dropped events that fail to be converted to json.
-#    # TODO(sissel): Find a way to continue passing events through even
-#    # if they fail to convert properly.
-#    begin
-#      payload = event.to_json
-#    rescue Encoding::UndefinedConversionError, ArgumentError
-#      puts "FAILUREENCODING"
-#      @logger.error("Failed to convert event to JSON. Invalid UTF-8, maybe?",
-#                    :event => event.inspect)
-#      return
-#    end
-#
-#    @producer.sendMsg(key,payload)
-#  end # def receive
-#
-#  public
-#  def teardown
-#    @kafka_client_queue.push(:stop_plugin)
-#  end
-#
-#  private
-#  def queue_event(msg, output_queue)
-#    begin
-#      @codec.decode(msg) do |event|
-#        decorate(event)
-#        event['kafka'] = {'msg_size' => msg.bytesize, 'topic' => @topic_id, 'consumer_group' => @group_id}
-#        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);
-#    end # begin
-#  end # def queue_event
-#
-#end #class LogStash::Inputs::Kafka
+require 'logstash/namespace'
+require 'logstash/outputs/base'
+require 'jruby-kafka'
+
+class LogStash::Outputs::Kafka < LogStash::Outputs::Base
+  config_name 'kafka'
+  milestone 1
+
+  default :codec, 'plain'
+
+  config :zk_connect, :validate => :string, :required => true
+  config :broker_list, :validate => :string, :required => true
+  config :topic_id, :validate => :string, :required => true
+
+  public
+  def register
+    jarpath = File.join(File.dirname(__FILE__), "../../../vendor/jar/kafka*.jar")
+    Dir[jarpath].each do |jar|
+      require jar
+    end
+    options = {
+        :zk_connect => @zk_connect,
+        :topic_id => @topic_id,
+        :broker_list => @broker_list
+    }
+    @producer = Kafka::Producer.new(options)
+    @producer.connect()
+    @logger.info('Registering kafka', :topic_id => @topic_id, :zk_connect => @zk_connect)
+  end # def register
+
+  public
+  def receive(event)
+    return unless output?(event)
+
+    key = event.sprintf(@key)
+    # TODO(sissel): We really should not drop an event, but historically
+    # we have dropped events that fail to be converted to json.
+    # TODO(sissel): Find a way to continue passing events through even
+    # if they fail to convert properly.
+    begin
+      payload = event.to_json
+    rescue Encoding::UndefinedConversionError, ArgumentError
+      puts "FAILUREENCODING"
+      @logger.error("Failed to convert event to JSON. Invalid UTF-8, maybe?",
+                    :event => event.inspect)
+      return
+    end
+
+    @producer.sendMsg(key,payload)
+  end # def receive 
+
+  public
+  def teardown
+    @kafka_client_queue.push(:stop_plugin)
+  end
+
+  private
+  def queue_event(msg, output_queue)
+    begin
+      @codec.decode(msg) do |event|
+        decorate(event)
+        event['kafka'] = {'msg_size' => msg.bytesize, 'topic' => @topic_id, 'consumer_group' => @group_id}
+        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);
+    end # begin
+  end # def queue_event
+
+end #class LogStash::Inputs::Kafka

adding producer back