input {
 file {
 path => "/tmp/access_log/*"
start_position => "beginning"
 }
}

filter {
 grok {
 match => { "message" => "%{COMBINEDAPACHELOG}" }
 }
 date {
 match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
 }
}

output {
 kafka {
 bootstrap_servers => "10.0.8.7:9092, 10.0.8.8:9092, 10.0.8.13:9092"
 topic_id => "apache-access-log"
 codec => plain {
 format => "%{message}"
 }
}
 stdout { codec => rubydebug }
}