KafkaConsumer无法读取主题全部记录的问题排查请求

我来帮你拆解下这个问题，你的核心问题出在消费者代码逻辑错误加上默认的Kafka拉取配置限制，跟主题大小限制没关系——Kafka主题只要磁盘空间足够，完全可以存储百万级甚至更多的消息。

问题原因详解 #

1. 强制重置offset的致命逻辑
   你在消费者的循环里每次都调用了consumer.seekToBeginning(consumer.assignment())，这会强制消费者每次都从分区的起始位置开始拉取消息。配合Kafka默认的拉取限制，每次只能拉取到有限数量的消息（你看到的778条），然后下次循环又回到开头，永远重复拉取这一小批消息，根本不会去读取后续的内容。

2. Kafka默认拉取配置的限制
   Kafka消费者有两个关键的默认配置会限制单次拉取的消息数：

   • max.partition.fetch.bytes：默认1MB，控制每个分区单次拉取的最大字节量。你的PMessage序列化后每条大概1.28KB左右，1MB刚好能装下约778条，这就是你看到这个数字的原因。
   • fetch.max.records：默认500，控制单次poll请求能拉取的最大记录数（不过这里你的情况主要是字节限制先触发了）。

3. 未手动提交offset
   你把ENABLE_AUTO_COMMIT_CONFIG设为了false，但代码里完全没有手动提交offset的逻辑。就算去掉了seekToBeginning，消费者重启后也会重新从起始位置拉取，不过当前问题里这是次要因素。

解决方案 #

1. 修正消费者代码逻辑

首先必须移除consumer.seekToBeginning(consumer.assignment())这行代码，然后添加手动提交offset的逻辑，确保消费者能正常推进offset。

2. 调整消费者拉取配置

根据你的消息量，调大相关拉取配置，让消费者单次能拉取到全部125000条消息。

修正后的消费者代码示例：

object ConsumerApp extends App {
  val topic = "topicTest"
  val properties = new Properties
  properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
  properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer")
  properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
  properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
  properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  
  // 调整拉取配置，适配你的消息量
  properties.put(ConsumerConfig.FETCH_MAX_RECORDS_CONFIG, "150000") // 单次最多拉取15万条
  properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "10485760") // 10MB，足够装下12.5万条1KB左右的消息

  val consumer = new KafkaConsumer[String, String](properties)
  consumer.subscribe(scala.List(topic).asJava)
  while (true) {
    val records: ConsumerRecords[String,String] = consumer.poll(Duration.ofMillis(20000))
    println("records size " + records.count())
    // 手动提交offset，确保下次从正确位置拉取
    consumer.commitSync()
  }
}

额外说明 #

• 如果你不需要重复消费，绝对不要随便调用seekToBeginning或seek方法，消费者默认会根据提交的offset自动推进。
• 调整拉取配置时要注意消费者的内存压力，如果单次拉取的消息字节量太大，可能会导致OOM，需要根据你的消费者服务内存情况合理设置。
• Kafka主题本身没有大小限制，只要你的Broker磁盘空间足够，就能存储任意多的消息。

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