gistfile1.txt

input {
    kafka {
        zk_connect => "10.4.22.121:2181,10.4.22.122:2181,10.4.22.123:2181"
        group_id => "appgroup"
        topic_id => "applog"
        #codec => plain
        consumer_threads => 64
        decorate_events => true
        reset_beginning => false
        queue_size => 2000
        #reset_beginning => true
        #auto_offset_reset => smallest
    }
}

filter {
     metrics {
          meter => "events"
          add_tag => "metric"
     }
}


output{
  if "metric" in [tags] {
        stdout {
          codec => line {
            format => "rate: %{[events][rate_1m]}"
          }
        }
      }
}