logstash config file for capturing kafka messages

logstash-kafka.conf

input {
    kafka {
        zk_connect => "172.17.0.215:2181"
        group_id => "logstash"
	topic_id => "request"
        reset_beginning => false
        consumer_threads => 1
        queue_size => 20
        rebalance_max_retries => 4
	rebalance_backoff_ms => 2000
	consumer_timeout_ms => -1
	consumer_restart_on_error => true
	consumer_restart_sleep_ms => 0
	codec => json
        }
}
input {
    kafka {
        zk_connect => "172.17.0.215:2181"
        group_id => "logstash"
	topic_id => "response"
        reset_beginning => false
        consumer_threads => 1
        queue_size => 20
        rebalance_max_retries => 4
	rebalance_backoff_ms => 2000
	consumer_timeout_ms => -1
	consumer_restart_on_error => true
	consumer_restart_sleep_ms => 0
	codec => json
        }
}

filter {
	if [response][compressed] {
	  ruby {
	  	code => "
			require 'zlib';
			temp_payload = event['response']['payload']
			#buf_payload = new_payload.pack('C*') if new_payload.is_a?(Array)
			#puts compressYN
			#puts new_payload.class
			event['response']['payload_decoded']= Zlib::Inflate.inflate(temp_payload.to_a.pack('C*'))
		"
		}
   	 }
}

output {
   stdout { codec => rubydebug }
	elasticsearch {
		protocol => "http"
		host => "localhost"
		index => "logs"
}
}