# How to filter kafka topics based on their names in logstash conf in the output section using if-else condition for elastic search output **URL:** \ **Category:** Logstash\ **Created:** [December 14, 2017, 9:30pm UTC](https://discuss.elastic.co/t/how-to-filter-kafka-topics-based-on-their-names-in-logstash-conf-in-the-output-section-using-if-else-condition-for-elastic-search-output/111863 "2017-12-14T21:30:54Z")\ **Posts on this page:** 3\ **Page:** 1
**Author:** ![KUMAR\_SATYAM](https://avatars.discourse-cdn.com/v4/letter/k/9d8465/32.png) [@KUMAR\_SATYAM](https://discuss.elastic.co/u/KUMAR_SATYAM)\ **Post date:** [December 14, 2017, 9:30pm UTC](https://discuss.elastic.co/t/how-to-filter-kafka-topics-based-on-their-names-in-logstash-conf-in-the-output-section-using-if-else-condition-for-elastic-search-output/111863/1 "2017-12-14T21:30:55Z")
I am using the new version of Kafka plugin for elkstack ``` input { kafka { bootstrap_servers => "1XX.X.X.X:9092" topics => ["test_all_logs","test_apiserver_logs","test_orchestrator_logs","test_credentialstore_logs","test_gfac_logs","local_api-orch_logs","__consumer_offsets,local_gfac_logs"] auto_offset_reset => "earliest" decorate_events => "true" } } filter{ json { source => "message" target => "doc" } mutate { add_field => { "level" => "%{[doc][level]}"}} mutate { add_field => { "logger" => "%{[doc][loggerName]}" } } } output { stdout { codec => rubydebug } elasticsearch { if {[@metadata][kafka][topic]== "test_all_logs" { hosts => ["localhost:9200"] index => ["test-all-logs-%{+YYYY.MM.dd}"] } if [@metadata][kafka][topic]} == "test_apiserver_logs" { hosts => ["localhost:9200"] index => ["test_apiserver_logs-%{+YYYY.MM.dd}"] } if [@metadata][kafka][topic]} == "test_orchestrator_logs" { hosts => ["localhost:9200"] index => ["test_orchestrator_logs-%{+YYYY.MM.dd}"] } if [@metadata][kafka][topic]} == "test_credentialstore_logs" { hosts => ["localhost:9200"] index => ["test_credentialstore_logs-%{+YYYY.MM.dd}"] } if [@metadata][kafka][topic]} == "test_gfac_logs" { hosts => ["localhost:9200"] index => ["test_gfac_logs-%{+YYYY.MM.dd}"] } } } ``` I am following many links which did not had proper answer: [https://stackoverflow.com/questions/41361756/how-to-write-logstash-filter-to-filter-kafka-topics](https://stackoverflow.com/questions/41361756/how-to-write-logstash-filter-to-filter-kafka-topics) [https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-decorate\_events](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-decorate_events) I got a suggestion that I should add `decorate_events => "true"`. and added @metadata attribute. I executed the command: ``` /opt/logstash/bin/logstash -f airavata/logstash-airavata.conf --path.data ./airavata/ ``` But I am getting the below error: ``` Failed to execute action {:action=>LogStash::PipelineAction::Create/pipeline_id:main, :exception=>"LogStash::ConfigurationError", :message=>"Expected one of #, => at line 24, column 8 (byte 579) after output {\n stdout { codec => rubydebug }\n\n elasticsearch {\n if ", :backtrace=>["/opt/logstash/logstash-core/lib/logstash/compiler.rb:42:in `compile_ast'", "/opt/logstash/logstash-core/lib/logstash/compiler.rb:50:in `compile_imperative'", "/opt/logstash/logstash-core/lib/logstash/compiler.rb:54:in `compile_graph'", "/opt/logstash/logstash-core/lib/logstash/compiler.rb:12:in `block in compile_sources'", "org/jruby/RubyArray.java:2486:in `map'", "/opt/logstash/logstash-core/lib/logstash/compiler.rb:11:in `compile_sources'", "/opt/logstash/logstash-core/lib/logstash/pipeline.rb:107:in `compile_lir'", "/opt/logstash/logstash-core/lib/logstash/pipeline.rb:49:in `initialize'", "/opt/logstash/logstash-core/lib/logstash/pipeline.rb:215:in `initialize'", "/opt/logstash/logstash-core/lib/logstash/pipeline_action/create.rb:35:in `execute'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:335:in `block in converge_state'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:141:in `with_pipelines'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:332:in `block in converge_state'", "org/jruby/RubyArray.java:1734:in `each'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:319:in `converge_state'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:166:in `block in converge_state_and_update'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:141:in `with_pipelines'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:164:in `converge_state_and_update'", "/opt/logstash/logstash-core/lib/logstash/agent.rb:90:in `execute'", "/opt/logstash/logstash-core/lib/logstash/runner.rb:362:in `block in execute'", "/opt/logstash/vendor/bundle/jruby/2.3.0/gems/stud-0.0.23/lib/stud/task.rb:24:in `block in initialize'"]} ``` Can somebody help me with mistake I am doing, with proper syntax, not uncooked syntax? ---
**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\ **Post date:** [December 15, 2017, 10:43am UTC](https://discuss.elastic.co/t/how-to-filter-kafka-topics-based-on-their-names-in-logstash-conf-in-the-output-section-using-if-else-condition-for-elastic-search-output/111863/2 "2017-12-15T10:43:59Z")
You can't have conditionals _inside_ outputs. Do this instead: ```nohighlight if (...) { elasticsearch { ... } } else if (...) { elasticsearch { ... } } ... ``` ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [January 12, 2018, 10:44am UTC](https://discuss.elastic.co/t/how-to-filter-kafka-topics-based-on-their-names-in-logstash-conf-in-the-output-section-using-if-else-condition-for-elastic-search-output/111863/3 "2018-01-12T10:44:01Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.