如何处理Flink KafkaSource的异常的数据

[Summer]

你好，我想问一下，如果来源于Kakfka的一条数据出现错误，会导致任务执行失败，日志抛出该条错误数据。


为保证任务执行，需要在*** WITH内加'value.json.ignore-parse-errors' = 'true',  
'value.json.fail-on-missing-field' = 'false'




那么之后如果出现异常的数据，我应该怎么感知到呢？？