使用Flink KafkaSource消费3分区Kafka主题为何遗漏消息？

云上老周

2026-7-1

问题排查与解决

1. 代码笔误导致未使用正确的Kafka Source

你的代码中存在明显的变量名错误：

val testSource = KafkaSource.builder[String]...build()
// 此处误用了未定义的tdiceSource，而非上面构建的testSource
ingestCommon.sEnv.fromSource(tdiceSource, WatermarkStrategy.noWatermarks(), "KafkaConsumer")

这会导致Flink实际运行的不是你配置的TestTopic订阅源，自然会遗漏消息。修正为：

ingestCommon.sEnv.fromSource(testSource, WatermarkStrategy.noWatermarks(), "KafkaConsumer")

2. 未开启Checkpoint导致故障重启丢消息

Flink Kafka Source依赖Checkpoint机制保证Exactly-Once语义，若未开启Checkpoint：

任务重启时会从已提交的偏移量位置恢复，未处理的消息不会被重新消费
偏移量无法被正确持久化，易出现丢消息情况

开启Checkpoint的示例代码：

ingestCommon.sEnv.enableCheckpointing(5000) // 每5秒触发一次Checkpoint
ingestCommon.sEnv.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

3. 偏移量配置可能导致的遗漏

你使用了OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)：

如果该消费组之前有提交过偏移量，任务启动时会从提交位置开始消费，提交位置之后的未消费消息不会被回溯
若需要从头开始消费，可调整为：

.setStartingOffsets(OffsetsInitializer.earliest())

4. 反序列化失败导致消息丢失

若消息格式与SimpleStringSchema不匹配，Flink默认会跳过错误消息。可添加错误处理逻辑排查：

.setValueOnlyDeserializer(new SimpleStringSchema() {
  override def deserialize(bytes: Array[Byte]): String = {
    try {
      super.deserialize(bytes)
    } catch {
      case e: Exception =>
        println(s"反序列化失败的消息：${new String(bytes)}")
        null
    }
  }
})

同时查看Flink日志是否存在反序列化相关异常。