如何处理卡夫卡接收器中的异常?

Question

我有一个将数据写入Kafka的Flink作业。Kafka 主题的最大消息大小设置为 5 MB，因此如果我尝试写入任何大于 5 MB 的记录，它会引发以下异常并导致作业关闭。

java.lang.Exception: Failed to send data to Kafka: The request included a message larger than the max message size the server will accept.
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.checkErroneous(FlinkKafkaProducerBase.java:373)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010.invoke(FlinkKafkaProducer010.java:350)
at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:56)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:549)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:524)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:504)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:830)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:808)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:207)
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:69)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:264)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.

现在，我已经在我的作业中配置了检查点，因此如果作业失败，它将再次重新启动。问题是，每次重新启动时，它都会针对同一记录失败，并进入故障和重新启动的无限循环。有没有办法在我的代码中处理这个 Kafka 异常，这样它就不会导致整个作业瘫痪？

Answer 1

也许你可以在 Kafka 接收器前面引入一个过滤器，它可以检测并过滤掉太大的记录。有点笨拙，但可能很容易。否则，我会考虑扩展 FlinkKafkaProducer010 以便能够处理异常。