# Creating Headers (with key value) in kafka offset using logstash Configuration

**URL:** <https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089>\
**Category:** Logstash\
**Created:** [June 15, 2023, 11:58am UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089 "2023-06-15T11:58:48Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![Ayushi\_bhardwaj](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/ayushi_bhardwaj/32/109915_2.png) [@Ayushi\_bhardwaj](https://discuss.elastic.co/u/Ayushi_bhardwaj)\
**Post date:** [June 15, 2023, 11:58am UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089/1 "2023-06-15T11:58:48Z")

</div>

Below is my config i had written to create headers (key & value) in my logstash config , but still it does not reflect to me in headers under offset tool.

```auto
input {
    elasticsearch {
    hosts => ["localhost"]
    index => "some_index*"
    query => '{ "sort" : [{"@timestamp" : {"order" : "asc"}}], "query": { "range": { "@timestamp": {"gte": "now-1m","lt": "now"} }}}'
    size => 1000
    docinfo => true
    schedule => "*/2 * * * *"
    user => "elasticsearch"
    password => "password"
   }
}
filter {
 mutate {
  add_field => {
          "[@metadata][header_key1]" => "Hello World"
                "[@metadata][header_key2]" => "abc"
}
}
}

output{
  kafka {
                bootstrap_servers => "kafka-server1:9096"
                topic_id => "INT_TEST"
                security_protocol => "SASL_SSL"
                sasl_mechanism => "SCRAM-SHA-512"
                ssl_endpoint_identification_algorithm => "https"
                sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='abc' password='password';"
                client_id => "CLUSTER"
                key_serializer => "org.apache.kafka.common.serialization.StringSerializer"
                value_serializer => "org.apache.kafka.common.serialization.StringSerializer"

               codec => json {}
        }}

```

Please help how we can configure headers for kafka topic in logstash configuration

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [June 15, 2023, 12:08pm UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089/2 "2023-06-15T12:08:34Z")

</div>

> [@Ayushi\_bhardwaj](#):
>
> ```auto
> filter {
> mutate {
> add_field => {
> "[@metadata][header_key1]" => "Hello World"
> "[@metadata][header_key2]" => "abc"
> }
> }
> }
> 
> ```

`@metadata` is a special field to be used in a Logstash pipeline only, they are not present in the output document.

If you want those fields in the output document you need to use other name.

---

<div class="post-metadata">

**Author:** ![Ayushi\_bhardwaj](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/ayushi_bhardwaj/32/109915_2.png) [@Ayushi\_bhardwaj](https://discuss.elastic.co/u/Ayushi_bhardwaj)\
**Post date:** [June 16, 2023, 6:40am UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089/3 "2023-06-16T06:40:48Z")

</div>

Could you please share some sample configuration to create header in kafka topic under my logstash config

---

<div class="post-metadata">

**Author:** ![Ayushi\_bhardwaj](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/ayushi_bhardwaj/32/109915_2.png) [@Ayushi\_bhardwaj](https://discuss.elastic.co/u/Ayushi_bhardwaj)\
**Post date:** [June 16, 2023, 11:52am UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089/4 "2023-06-16T11:52:46Z")

</div>

Note:- We are using logstash-integration-kafka (10.12.0-java) plugin. Still we are not able create or write payload/value to Headers.

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [June 16, 2023, 12:54pm UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089/5 "2023-06-16T12:54:28Z")

</div>

> [@Ayushi\_bhardwaj](#):
>
> Could you please share some sample configuration to create header in kafka topic under my logstash config

I don't think this is supported, there is nothing specific about headers in the [kafka output documentation](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-kafka.html).

And there is this [old github issue](https://github.com/logstash-plugins/logstash-output-kafka/issues/224) asking for it to be implemented, but the issue is still open.

Apparently you can [read headers](https://github.com/logstash-plugins/logstash-integration-kafka/pull/78) but not write it.

---

<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:** [July 14, 2023, 12:55pm UTC](https://discuss.elastic.co/t/creating-headers-with-key-value-in-kafka-offset-using-logstash-configuration/336089/6 "2023-07-14T12:55:16Z")

</div>

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