# Kafka input to logstash

**URL:** <https://discuss.elastic.co/t/kafka-input-to-logstash/190150>\
**Category:** Logstash\
**Created:** [July 12, 2019, 7:07am UTC](https://discuss.elastic.co/t/kafka-input-to-logstash/190150 "2019-07-12T07:07:16Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Harsha606](https://avatars.discourse-cdn.com/v4/letter/h/6a8cbe/32.png) [@Harsha606](https://discuss.elastic.co/u/Harsha606)\
**Post date:** [July 12, 2019, 7:07am UTC](https://discuss.elastic.co/t/kafka-input-to-logstash/190150/1 "2019-07-12T07:07:16Z")

</div>

Hi All,

i am trying to add multiple kafka topics to logstash and output to elastic. my requirement here is, the index name should be the topic name, as below...

```auto
input {
  kafka {
    bootstrap_servers => "broker.kafka-elk.l4lb.thisdcos.directory:9092"
    topics => ["cid-test-express","cid-test-default"]
    group_id => "test-consumer-group"
# auto_offset_reset => "earliest"
    decorate_events => true
    codec => json
}
}
filter {
grok {
        match => { "[log][file][path]" => "/mnt/volume0/cid-v[0-9]-.*-service/services/(?<version>[^/]+)/" }
}
}

output {
if [@metadata][topic] == "cid-test-express" and [fields][log_type] == "express" {
elasticsearch {
hosts => ["es-master1.cidaas-elk.l4lb.thisdcos.directory:9200","es-master2.cidaas-elk.l4lb.thisdcos.directory:9200","es-master3.cidaas-elk.l4lb.thisdcos.directory:9200"]
index => "%{[@metadata][kafka][topic]}-%{version}-%{+YYYY.MM.dd}"
}
}
if [@metadata][topic] == "cid-test-default" and [fields][log_type] == "default" {
elasticsearch {
hosts => ["es-master1.cidaas-elk.l4lb.thisdcos.directory:9200","es-master2.cidaas-elk.l4lb.thisdcos.directory:9200","es-master3.cidaas-elk.l4lb.thisdcos.directory:9200"]
index => "%{[@metadata][kafka][topic]}-%{version}-%{+YYYY.MM.dd}"
}
}
}

```

only first if statement works, but not the second one, second topic index is not created!!! can any one help me regarding this!!! thanks in advance

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [August 9, 2019, 7:07am UTC](https://discuss.elastic.co/t/kafka-input-to-logstash/190150/2 "2019-08-09T07:07:19Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.