logstash-kafka

1. Introduction
   1. This project implements Kafka 0.8.1.1 inputs and outputs for logstash
2. Installation
   1. wget https://github.com/joekiller/logstash-kafka/archive/master.zip
   2. unzip master
   3. cd logstash-kafka-master
   4. make tarball LOGSTASH_VERSION=1.4.2
   5. cd build/tarball
   6. ls
3. Input
   

   input {
       kafka {
           zk_connect => ... # string (optional), default: "localhost:2181"
           group_id => ... # string (optional), default: "logstash"
           topic_id => ... # string (optional), default: "test"
           reset_beginning => ... # boolean (optional), default: false
           consumer_threads => ... # number (optional), default: 1
           queue_size => ... # number (optional), default: 20
           rebalance_max_retries => ... # number (optional), default: 4
           rebalance_backoff_ms => ... # number (optional), default:  2000
           consumer_timeout_ms => ... # number (optional), default: -1
           consumer_restart_on_error => ... # boolean (optional), default: true
           consumer_restart_sleep_ms => ... # number (optional), default: 0
           decorate_events => ... # boolean (optional), default: true
           consumer_id => ... # string (optional) default: nil
           fetch_message_max_bytes => ... # number (optional) default: 1048576
       }
   }

4. Output
   

   output {
       kafka {
           broker_list => ... # string (optional), default: "localhost:9092"
           topic_id => ... # string (optional), default: "test"
           compression_codec => ... # string (optional), one of ["none", "gzip", "snappy"], default: "none"
           compressed_topics => ... # string (optional), default: ""
           request_required_acks => ... # number (optional), one of [-1, 0, 1], default: 0
           serializer_class => ... # string, (optional) default: "kafka.serializer.StringEncoder"
           partitioner_class => ... # string (optional) default: "kafka.producer.DefaultPartitioner"
           request_timeout_ms => ... # number (optional) default: 10000
           producer_type => ... # string (optional), one of ["sync", "async"] default => 'sync'
           key_serializer_class => ... # string (optional) default: nil
           message_send_max_retries => ... # number (optional) default: 3
           retry_backoff_ms => ... # number (optional) default: 100
           topic_metadata_refresh_interval_ms => ... # number (optional) default: 600 * 1000
           queue_buffering_max_ms => ... # number (optional) default: 5000
           queue_buffering_max_messages => ... # number (optional) default: 10000
           queue_enqueue_timeout_ms => ... # number (optional) default: -1
           batch_num_messages => ... # number (optional) default: 200
           send_buffer_bytes => ... # number (optional) default: 100 * 1024
           client_id => ... # string (optional) default: ""
       }
   }

5. Reference
   1. https://github.com/joekiller/logstash-kafka