Debezium encodes my headers in base64

[Krzysztof Kasica]

Hi,
I'm developing a custom SMT for my connector that extracts values from an db column and sets them as headers. While making an integration test I've noticed that each of my headers values are encoded in BASE64.

I'm also using EventRouter SMT and the id header that's added is also encoded in BASE64 so the problem is not exclusive to my SMT.

Is there a setting in configuration that I can use to disable it?

I'm using a MySQL debezium connector version 2.4.

Krzysztof

[Krzysztof Kasica]

[jiri.p...@gmail.com]

Hi,

what is the source column datatype and what connector do you use?

Jiri

[Krzysztof Kasica]

The source column datatype for the headers column is json and the id column, which also ends up encoded, is of VARBINARY datatype.

I'm using Debezium MySQL connector version 2.4 with Confluent connect worker 'confluentinc/cp-server-connect' version 7.7.1.

[jiri.p...@gmail.com]

What is your setting of https://debezium.io/documentation/reference/3.0/connectors/mysql.html#mysql-property-binary-handling-mode ?
What do you use as header converter?

Jiri

[Krzysztof Kasica]

1. I'm using the default value of binary.handling.mode, bytes. I tried setting it explicitly to bytes and even base64 but the results are the same.

2. I'm not sure what you mean by header converter. I haven't configured anything like that. The only thing that is setting my headers is EventRouter SMT(for the id header) and my custom SMT.

Krzysztof

[jiri.p...@gmail.com]

Please check https://kafka.apache.org/documentation/#connectconfigs_header.converter

[Krzysztof Kasica]

I changed the header.converter to io.debezium.converters.BinaryDataConverter and it worked!

I've looked at the source code for SimpleHeaderConverter which is the one used by default. 'fromConnectHeader' method returns Values.convertToString(...). Inside Values.convertToString() you can find a function 'append' being called and finally inside 'append' if the value is a byte array it gets encoded in base64.

Link to 'append' function: https://github.com/apache/kafka/blob/trunk/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java#L712

Thank you for setting me on the right path Jiri!