Hi all, Is there a way to handle hafka deserialization exceptions, when a JSON message is malformed for example?
I thought about extending the DeserializationSchema to emit a null or any other value, but that may cause an NPE when using a subsequent TimestampExtractor. The other solution would be to do the serialization in a subsequent flatMap operator but it would be more convient to do it directly in the consumer to make the timestamps extraction (with the new kafka consumer timestamp extractor) easier. Any suggestions? Best, Yassine
