Flink1.15反序列化Kafka消息返回null无法跳过坏消息重复消费问题

1. 版本语义变更：Flink 1.14版本对Kafka连接器和反序列化体系做了重构，AbstractDeserializationSchema.deserialize()方法返回null的语义从1.13及更早版本的「跳过当前坏消息」，改成了「标识当前数据流到达末尾」，对应的文档说明也在1.14版本同步移除，旧的隐式处理逻辑没有做向后兼容。
2. 重复消费触发逻辑：1.15版本中，反序列化方法返回null既不会触发消息过滤，也不会提交对应消息的消费位移，Flink会判定这条消息未被成功处理，在后续的分区拉取轮询、任务重启场景下会重新拉取该消息，最终形成坏消息无限重复消费的循环。调试可确认代码进入异常分支返回null、但消息持续被消费的现象，和这个版本的逻辑完全一致，不属于业务分支判断的代码错误。
3. 额外说明：1.15版本中旧版FlinkKafkaConsumer已经被标记为废弃不再维护，该组件内部对反序列化null返回值的处理存在已知bug，无修复计划。

不要依赖返回null实现坏消息跳过，根据任务场景选以下任意一种方案即可：

• 方案1：生产环境推荐做法，使用侧输出流收集脏数据
  实现KafkaDeserializationSchema<T>接口替代继承AbstractDeserializationSchema，在deserialize方法中捕获XML解析异常后，不要返回null，将解析失败的坏消息写入预先定义的侧输出流，主流只输出解析成功的业务数据。侧输出流后续可以接脏数据告警、审计落库等逻辑，消息会被正常消费提交位移，不会出现重复拉取。
• 方案2：快速适配，过滤无效标记对象
  在捕获XML解析异常后，返回一个自定义的全局唯一无效消息实例，在Kafka Source之后接第一个算子时，直接过滤掉这类无效实例即可，逻辑简单改动量小，适合快速修复线上问题。
• 方案3：迁移到FLIP-27标准的新KafkaSource，配置官方脏数据跳过策略
  废弃旧版FlinkKafkaConsumer，使用重构后的KafkaSource组件，构建Source时显式开启反序列化错误忽略配置，代码示例如下：

  KafkaSource<YourBusinessClass> kafkaSource = KafkaSource.<YourBusinessClass>builder()
          .setBootstrapServers("your-kafka-broker:9092")
          .setTopics("your-target-topic")
          .setGroupId("your-consumer-group")
          .setStartingOffsets(OffsetsInitializer.latest())
          .setDeserializer(KafkaRecordDeserializationSchema.of(new YourCustomXMLDeserializationSchema()))
          // 开启反序列化异常跳过，坏消息直接丢弃不重试
          .setProperty("kafka.source.deserialization.ignore-parse-errors", "true")
          .build();
  

  开启该配置后，反序列化阶段抛出的所有异常都会被Source层捕获，对应坏消息直接跳过、位移正常提交，如果需要收集脏数据做审计，还可以搭配官方脏数据收集器将坏消息输出到指定存储。

相关参考材料：

• 自定义反序列化代码实现：
• 1.13版本文档规则说明：
• 坏消息触发重复消费运行流程：

内容的提问来源于stack exchange，提问作者Koman