Logstash kafka benchmark

logstash.conf

input {
  kafka {
    topic_id => 'observations.json'
    zk_connect => 'localhost:2181'
    type => 'kafka'
    auto_offset_reset => 'smallest'
    consumer_threads => 12
   codec => 'snappy'
  }
}

filter {
  if [type] == "kafka" {
    metrics {
      meter => "events"
      add_tag => "metric"
    }
  }
}

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