# Kafka input plugin does not set @metadata fields

**URL:** <https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256>\
**Category:** Logstash\
**Created:** [March 22, 2018, 7:30pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256 "2018-03-22T19:30:17Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![jeffkirk1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jeffkirk1/32/8080_2.png) [@jeffkirk1](https://discuss.elastic.co/u/jeffkirk1)\
**Post date:** [March 22, 2018, 7:30pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256/1 "2018-03-22T19:30:17Z")

</div>

The kafka input plugin does not appear to set the @metadata fields as described in this [Logstash documentation](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html).

I'm using Elasticsearch 6.2.2 and Logstash 6.2.2 on Docker images, running on a CentOS 7.4 server.

I wanted to set up a kafka input pipeline that would read from any topic defined in an array. Based on the documentation, I should have been able to set the index name used for Elasticsearch in the output plugin using this syntax:

```
output {
  elasticsearch {
    hosts => <list of hosts redacted>
    index => "%{[@metadata][kafka][topic]}-%{+YYYY.MM.dd}"
  }
}

```

Instead, the index pattern generated by data sent to any of the topics is set to the literal string, %{[@metadata][kafka][topic]}- followed by the date. I now have an index in my Elasticsearch cluster for today named %{[@metadata][kafka][topic]}-2018.03.22.

To test whether %{foo} substitution was working in the output plugin, I created three kafka input instances, one for each of the topics I wanted to read from, and added a field named "topic" with the value set to an Elasticsearch-compatible string version of the topic. This works: I get three indexes with the current date substituted in properly: test-YYYY.MM.dd, sensor\_events-YYYY.MM.dd, and controller\_log-YYYY.MM.dd.

Here's the workaround config:

```
input {

  kafka {
    bootstrap_servers => <kafka_server_list_redacted>
    topics => ["test"]
    consumer_threads => 3
    auto_offset_reset => "latest"
    auto_commit_interval_ms => "500"
    enable_auto_commit => true
    codec => json {
      charset => "ISO-8859-1"
    }
    add_field => { "topic" => "test" }
  }

  kafka {
    bootstrap_servers => <kafka_server_list_redacted>
    topics => ["sensorEvents"]
    consumer_threads => 3
    auto_offset_reset => "latest"
    auto_commit_interval_ms => "500"
    enable_auto_commit => true
    codec => json {
      charset => "ISO-8859-1"
    }
    add_field => { "topic" => "sensor_events" }
  }

  kafka {
    bootstrap_servers => <kafka_server_list_redacted>
    topics => ["controllerLog"]
    consumer_threads => 3
    auto_offset_reset => "latest"
    auto_commit_interval_ms => "500"
    enable_auto_commit => true
    codec => json {
      charset => "ISO-8859-1"
    }
    add_field => { "topic" => "controller_log" }
  }
}

output {
  elasticsearch {
    hosts => <elasticsearch_hosts_redacted>
    index => "%{[topic]}-%{+YYYY.MM.dd}"
  }
}

```

I also tried creating a mutate filter to set the "topic" field to the contents of the [@metadata][kafka][topic] variable, and this doesn't work either. I can only conclude that the [@metadata][kafka][topic] is never populated by the plugin.

---

<div class="post-metadata">

**Author:** ![yaauie](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/yaauie/32/23363_2.png) [@yaauie](https://discuss.elastic.co/u/yaauie)\
**Post date:** [March 22, 2018, 8:19pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256/2 "2018-03-22T20:19:41Z")

</div>

Oy. That's frustrating.

Metadata is only added to the event if the `decorate_events` option is set to `true` (it defaults to `false`); we should definitely make it easier to find and make sense of what's included 😩.

> `decorate_events`
> 
> - Value type is boolean
> - Default value is false
> 
> Option to add Kafka metadata like topic, message size to the event.
> 
> - [Kafka input plugin | Logstash Reference [8.11] | Elastic](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-decorate_events)

I've opened an issue on the plugin to improve the docs: [logstash-plugins/logstash-input-kafka#255](https://github.com/logstash-plugins/logstash-input-kafka/issues/255)

---

<div class="post-metadata">

**Author:** ![jeffkirk1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jeffkirk1/32/8080_2.png) [@jeffkirk1](https://discuss.elastic.co/u/jeffkirk1)\
**Post date:** [March 22, 2018, 8:39pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256/3 "2018-03-22T20:39:28Z")

</div>

Thanks! That would help. 😃

---

<div class="post-metadata">

**Author:** ![jeffkirk1](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jeffkirk1/32/8080_2.png) [@jeffkirk1](https://discuss.elastic.co/u/jeffkirk1)\
**Post date:** [March 22, 2018, 9:16pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256/4 "2018-03-22T21:16:12Z")

</div>

As you might expect, that was clearly the issue. I thought I'd post the optimized config file for others to look at, as well as my solution for remapping the camel-case index names to something Elasticsearch can actually index. It throws an error when the index name contains capital letters.

Here's my solution that solved both problems:

```
input {

  kafka {
    bootstrap_servers => <kafka_server_list_redacted>
    topics => ["test", "controllerLog"]
    consumer_threads => 6
    auto_offset_reset => "latest"
    auto_commit_interval_ms => "500"
    enable_auto_commit => true
    decorate_events => true
    codec => json {
      charset => "ISO-8859-1"
    }
  }

}

filter {
  ruby {
    code => "event.set('[@metadata][kafka][lc_topic]', event.get('[@metadata][kafka][topic]').split(/(?=[A-Z])/).map{|x| x.downcase }.join('_') )"
  }
}

output {
  elasticsearch {
    hosts => <elasticsearch_hosts_redacted>
    index => "%{[@metadata][kafka][lc_topic]}-%{+YYYY.MM.dd}"
  }
}
```

---

<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:** [April 19, 2018, 9:16pm UTC](https://discuss.elastic.co/t/kafka-input-plugin-does-not-set-metadata-fields/125256/5 "2018-04-19T21:16:21Z")

</div>

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