ksqlDB Kafka header manipulation

[feynur]

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

[mjsax]

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.

[feynur]

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’
);

[mjsax]

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