[jira] [Created] (FLINK-18017) improve Kafka connector to handle record deserialization exception and report related metrics

["Yu Yang (Jira)" <ji...@apache.org>]

Yu Yang created FLINK-18017:
-------------------------------

             Summary: improve Kafka connector to handle record deserialization exception and report related metrics
                 Key: FLINK-18017
                 URL: https://issues.apache.org/jira/browse/FLINK-18017
             Project: Flink
          Issue Type: Improvement
          Components: Connectors / Kafka
    Affects Versions: 1.9.1
            Reporter: Yu Yang


Corrupted messages can get into the message pipeline for various reasons.  When a Flink deserializer fails to deserialize the message, and throw an exception due to corrupted message, the flink application will be blocked until we update the deserializer to handle the exception. 

 

Currently messages are deserialized as below in 

flink_pinterest/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaFetcher.java
{code:java}

for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {

 final T value = deserializer.deserialize(record);

 if (deserializer.isEndOfStream(value)) {
 // end of stream signaled
 running = false;
 break;
 }

 // emit the actual record. this also updates offset state atomically
 // and deals with timestamps and watermark generation
 emitRecord(value, partition, record.offset(), record);
}
  {code}
Flink Kafka connector needs to catch exception from deserialization, and expose related metrics. 

 



--
This message was sent by Atlassian Jira
(v8.3.4#803005)