[FLINK-18017] have Kafka connector report metrics on null records

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.9.1
• Component/s: Connectors / Kafka
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Yu Yang
• Created: 29/May/20 00:32
• Updated: 14/Nov/21 10:38

Description

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.  AbstractFetcher.emitRecordsWithTimestamps skips null records.  We need to add an metric on # of null records so that the users can measure # of null records that KafkaConnector encounters, and set up monitoring & alerting based on that. 

https://github.com/apache/flink/blob/1cd696d92c3e088a5bd8e5e11b54aacf46e92ae8/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java#L350