# How to save Kafka Headers values in MongoDB Sink? **URL:** \ **Category:** Self-Managed Connectors\ **Created:** [8 November 2022 10:11 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460 "2022-11-08T10:11:56Z")\ **Posts on this page:** 8\ **Page:** 1
**Author:** ![zikria35](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/zikria35/32/2460_2.png) [@zikria35](https://forum.confluent.io/u/zikria35)\ **Post date:** [8 November 2022 10:11 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/1 "2022-11-08T10:11:57Z")
Right now I have a MongoDB Sink connector. My incoming Kafka messages contain Auditlogs in the Kafka headers. The Kafka headers values are byte arrays and the Kafka message values are Avro. How can I best save the Header values (decoded to JSON) to my MongoDB sink? ---
**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\ **Post date:** [8 November 2022 22:57 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/2 "2022-11-08T22:57:49Z")
Have you tried to use a transform? > **[HeaderFrom | Confluent Documentation](https://docs.confluent.io/platform/current/connect/transforms/headerfrom.html#properties)** > > This document provides usage information for the Apache Kafka SMT org.apache.kafka.connect.transforms.HeaderFrom. ---
**Author:** ![zikria35](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/zikria35/32/2460_2.png) [@zikria35](https://forum.confluent.io/u/zikria35)\ **Post date:** [9 November 2022 08:13 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/3 "2022-11-09T08:13:18Z")
@OneCricketeer Yeah, sadly there is no transform for my use case that I know of. Headerfrom adds headers from values and I need it the other way around. ---
**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\ **Post date:** [9 November 2022 13:22 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/4 "2022-11-09T13:22:32Z")
What about this one? [https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/HeaderToField.html](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/HeaderToField.html) ---
**Author:** ![zikria35](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/zikria35/32/2460_2.png) [@zikria35](https://forum.confluent.io/u/zikria35)\ **Post date:** [9 November 2022 14:16 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/5 "2022-11-09T14:16:40Z")
@OneCricketeer Yes, I have used this. only problem is that it is saving the values as Base64 in the mongo sink. How can I save it as JSON? ---
**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\ **Post date:** [9 November 2022 15:01 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/6 "2022-11-09T15:01:16Z")
There’s a `BytesToString` transform in the same project. Otherwise, you should be able to decode your data at query time. ---
**Author:** ![zikria35](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/zikria35/32/2460_2.png) [@zikria35](https://forum.confluent.io/u/zikria35)\ **Post date:** [9 November 2022 15:04 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/7 "2022-11-09T15:04:16Z")
@OneCricketeer Hmm I am going to try this. During querying I dont think it will be very logical in case of filtering for example. I was also thinking of using a custom post processor: [https://www.mongodb.com/docs/kafka-connector/current/sink-connector/fundamentals/post-processors/](https://www.mongodb.com/docs/kafka-connector/current/sink-connector/fundamentals/post-processors/) that casts bytes to json ---
**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\ **Post date:** [24 November 2022 11:39 UTC](https://forum.confluent.io/t/how-to-save-kafka-headers-values-in-mongodb-sink/6460/8 "2022-11-24T11:39:50Z")
This topic was automatically closed 7 days after the last reply. New replies are no longer allowed.