Hi,
I'm using a custom KafkaSerializationSchema to write records to Kafka using FlinkKafkaProducer. The objects written are Rows coming from Flink's SQL API.
In some cases, when trying to convert the Row object to a byte[], serialization will fail due to malformed values. In such cases, I would like the custom serialization schema to drop the bad records and not send them through.
From the API, it is unclear how such failures should be handled. Given the following signature:
ProducerRecord<byte[], byte[]> serialize(T element, @Nullable Long timestamp);
From reading the code, there's no exception handling or null checking, which means that:
- If an exception is thrown, it will cause the entire job to fail (this has happened to me in production)
- If null is passed, a null value will be pushed down to kafkaProducer.send which is undesirable.
What are the options here?
--
Best Regards,
Yuval Itzchakov.