假如flink消费kafka消息，这条消息由于格式处理不了报异常了，这时候就没法ack给kafka[阿里云实时计算 Flink版]

大佬们，假如flink消费kafka消息，这条消息由于格式处理不了报异常了，这时候就没法ack给kafka，flink异常重启然后再次拉取数据，还是这一条错误数据，那不就卡死这里了，有什么解决办法吗？