[jira] [Updated] (FLINK-18017) have Kafka connector report metrics on null records

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

     [ https://issues.apache.org/jira/browse/FLINK-18017?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Yu Yang updated FLINK-18017:
----------------------------
    Summary: have Kafka connector report metrics on null records   (was: improve Kafka connector to handle record deserialization exception and report related metrics)

> have Kafka connector report metrics on null records 
> ----------------------------------------------------
>
>                 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
>            Priority: Major
>
> 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)