Hi all. I use

KafkaSink

to save my events in kafka, but during big load I getting flink app restart due to

Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 6520160 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.

what is pretty strange, since my messages I save via kafkaSink are json string with 2k char length maximum. not `6520160`bytes (6.5Mb). Pls point me to some documentation, what am I missing here? p.s. flink 1.14.6

I believe that the issue is on Kafka level and not related to Flink. I am pretty sure that your

message.max.bytes

configuration on broker side (or only for topic config) is

6520160

bytes. Apparently with spike of load you get an exceptionally big message. (kafka doc: https://kafka.apache.org/documentation/#brokerconfigs_message.max.bytes)

I dont want to have big

max.message.bytes

. Since I reach some limit but my messages much less than 1mb I think there’s some batching process. Can I control it via KafkaSink ?

it should be on the level of your application logic I believe. Apparently your job at the end of the day serialise too big message

no, I have a fairly static datastream of small messages (~1k msgs/s) that I try to write to kafka

frankly, your exception is quite transparent and deterministic, see the description: https://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/common/errors/RecordTooLargeException.html. I would track the size of the message and/or catch this exception in the sink and dump the message causing it

track the size of the message and/or catch this exception in the sink and dump the message causing it

btw, any how-to do that? this is what I looking for. thanks in advance

1. for exception handling: https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka[…]ain/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java you want to try catch this method I think (you can double check since you have the stacktrace) 2. for size tracking (the simplest way in non-prod env is to check the size of Seralized message and log when it is more than you expect is small) For prod it would be a little bit more tricky, since you want to emmit a new

gauge

metric probably

Many thanks @Oleksandr Nitavskyi p.s. привіт з Києва 🇺🇦

Hope you are fine there! 🇺🇦