Logstash configuration file for ingesting Stack Exchange events data from Kafka topic to Elasticsearch index

logstash.conf

input {
        kafka {
                bootstrap_servers => "KAFKA_ADDRESS_AND_PORT"
                group_id => "logstash"
                topics => ["posts"]
                codec => "json"
                type => "posts"
                consumer_threads => 2
        }
}

filter {
        mutate {
                add_field => { "EventDate" => "%{@timestamp}" }
        }
}

output {
        if [type] == "posts" {
                if [answer_id] {
                        elasticsearch {
                                hosts => ["ES_ADDRESS_AND_PORT"]
                                manage_template => false
                                index => "posts"
                                document_type => "answer"
                                document_id => "%{answer_id}"
                        }
                } else if [comment_id] {
                        elasticsearch {
                                hosts => ["ES_ADDRESS_AND_PORT"]
                                manage_template => false
                                index => "posts"
                                document_type => "comment"
                                document_id => "%{comment_id}"
                        }

                } else if [question_id] {
                        elasticsearch {
                                hosts => ["ES_ADDRESS_AND_PORT"]
                                manage_template => false
                                index => "posts"
                                document_type => "question"
                                document_id => "%{question_id}"
                        }
                }
        }
}