# ksqlDB Kafka header manipulation

**URL:** <https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475>\
**Category:** ksqlDB\
**Created:** [6 July 2022 06:37 UTC](https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475 "2022-07-06T06:37:38Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![feynur](https://avatars.discourse-cdn.com/v4/letter/f/74df32/32.png) [@feynur](https://forum.confluent.io/u/feynur)\
**Post date:** [6 July 2022 06:37 UTC](https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475/1 "2022-07-06T06:37:38Z")

</div>

Can I use ksqlDB to insert new header using original message values?

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [6 July 2022 19:45 UTC](https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475/2 "2022-07-06T19:45:28Z")

</div>

No. ksqlDB only gives you read-only access to headers atm. It’s not possible to modify or set headers for the output.

Of course, he hope to support header manipulation in the future.

Not sure what your overall requirements are, but if using Kafka Streams is an option for you, I would recommend to use it for this use case.

---

<div class="post-metadata">

**Author:** ![feynur](https://avatars.discourse-cdn.com/v4/letter/f/74df32/32.png) [@feynur](https://forum.confluent.io/u/feynur)\
**Post date:** [7 July 2022 05:37 UTC](https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475/3 "2022-07-07T05:37:34Z")

</div>

Thanks for the answer mjsax!  
How can I use kafka streams to manipulate the header?  
I am using source connector to publish messages to Kafka topic. In this topic I don’t have any header. Imagine that I have 3 field in the kafka message that are id, type and status. I would like to create new topic using TEST stream but also I would like to add **type** value to kafka header([{ “key”: “type”,“stringValue”: “tesytType”}]

CREATE STREAM TEST(  
id VARCHAR,  
type VARCHAR,  
status VARCHAR  
)WITH (  
KAFKA\_TOPIC = ‘test-topic’,  
VALUE\_FORMAT = ‘JSON’  
);

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [7 July 2022 20:11 UTC](https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475/4 "2022-07-07T20:11:23Z")

</div>

In Kafka Streams, you can use a `api.Processor` (note the package name `api.` – the old `Processor` is different).

It allows you to `context.forward(new Record().withHeader(...))`.

Check out the docs for details: [Kafka Streams Processor API | Confluent Documentation](https://docs.confluent.io/platform/current/streams/developer-guide/processor-api.html)

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [5 August 2022 06:38 UTC](https://forum.confluent.io/t/ksqldb-kafka-header-manipulation/5475/5 "2022-08-05T06:38:35Z")

</div>

This topic was automatically closed after 30 days. New replies are no longer allowed.