# Logstash kafka input plugin mutate problem

**URL:** <https://discuss.elastic.co/t/logstash-kafka-input-plugin-mutate-problem/284156>\
**Category:** Logstash\
**Created:** [September 14, 2021, 8:40am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-mutate-problem/284156 "2021-09-14T08:40:30Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![d.silwon](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/d.silwon/32/65853_2.png) [@d.silwon](https://discuss.elastic.co/u/d.silwon)\
**Post date:** [September 14, 2021, 8:40am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-mutate-problem/284156/1 "2021-09-14T08:40:30Z")

</div>

Dears,

I'm trying to load data from kafka topic but have some problem with if statement in filter.  
This part of the setup doesn't work and I don't know why:

```auto
replace => { "[@metadata][index_prefix]" => "kafka-%{+YYYY.MM.dd}" }

```

All messages from Kafka are loading to index called other-\*

This is my almost whole config of logstash:

```auto
input {
  kafka {
    bootstrap_servers => "10.10.131.18:9093"
    type => "kafka_type"
    topics => "TEST_ELK_TOPICS"
    decorate_events => true
    codec => "json"
    # ssl => true
    ssl_truststore_location => "/etc/logstash/certs/KafkaTruststore.p12"
    ssl_truststore_password => "elasticpwd"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='elastic' password='elastic123';"
    sasl_mechanism => "SCRAM-SHA-512"
    security_protocol => "SASL_SSL"
 }
}

filter {
  if "tomcat" not in [tags] or "app" not in [tags] {
  xml {
    source => "message"
    store_xml => false
    force_array => false
    xpath => [
      "/log//isomsg/field[@id='37']/@value", "hi.rrn",
      "/log//isomsg/field[@id='0']/@value", "hi.mti",
      "/log//isomsg/field[@id='39']/@value", "hi.rc",
      "/log//*[contains(name(),'exception')]/@name", "hi.exception_name",
      "/log//*[contains(name(),'exception')]/text()", "hi.exception",
      "/log//error/text()", "hi.error",
      "/log/@realm", "hi.realm",
      "/log/@at", "hi.xml_csshi_date",
      "/log//routing/dst-iface/text()", "hi.dst-iface"
    ]
  }
  }

  if "kafka_type" in [type] {
    json {
      source => "message"
    }
    mutate {
      add_field => { "kafka" => "%{[@metadata][kafka]}" }
      replace => { "[@metadata][index_prefix]" => "kafka-%{+YYYY.MM.dd}" }
      replace => { "[@metadata][_id]" => "%{[@metadata][kafka][offset]}%{[@metadata][kafka][timestamp]}" }
    }
  }

....
else if "webbapisrv" in [host][hostname] {
    mutate {
      replace => { "[@metadata][index_prefix]" => "webapi-%{+YYYY.MM.dd}" }
    }
  }
  else {
    mutate {
      replace => { "[@metadata][index_prefix]" => "other-%{+YYYY.MM.dd}" }
    }
  }
}

output {
  elasticsearch {
    hosts => ["https://${HOSTNAME}:9200"]
    cacert => '/etc/logstash/certs/ca.crt'
    user => 'logstash_internal'
    password => '${ES_PWD}'
    ilm_enabled => false
    document_id => "%{[@metadata][_id]}"
    index => "%{[@metadata][index_prefix]}"
  }
}

```

Could you advise me what is wrong, please? Thanks

Best Regards,  
Dan

---

<div class="post-metadata">

**Author:** ![d.silwon](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/d.silwon/32/65853_2.png) [@d.silwon](https://discuss.elastic.co/u/d.silwon)\
**Post date:** [September 14, 2021, 10:15am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-mutate-problem/284156/2 "2021-09-14T10:15:16Z")

</div>

A problematic condition was placed in bad place in configuration. The correction of config file solved the issue. Thanks.

---

<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:** [October 12, 2021, 10:16am UTC](https://discuss.elastic.co/t/logstash-kafka-input-plugin-mutate-problem/284156/3 "2021-10-12T10:16:12Z")

</div>

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