Spring Boot中Kafka单消费者无法稳定读取全量消息求助

问题根源 #

你的代码存在两个关键问题，导致无法稳定获取全部消息：

1. 分区分配未完成就执行seek操作：第一次poll(Duration.ofSeconds(1))的时间窗口太短，消费者可能还没完成与协调者的分区分配流程，此时consumer.assignment()拿到的只是部分分区，后续seekToBeginning只能针对这些分区生效，未被分配的分区会按照默认偏移量策略消费（可能不是从头开始）。
2. 单次poll无法保证拉取全部消息：Kafka的poll方法是批量拉取当前broker可用的消息，受fetch.max.records（默认500）、broker的消息批次策略、网络延迟等影响，单次调用不一定能拉取到所有消息。

修正方案 #

1. 确保分区完全分配后再执行seek #

需要循环轮询直到拿到所有分区，避免因分区未分配导致的seek不完整。

2. 循环poll收集所有消息 #

单次poll可能无法拉取全部消息，需要循环调用直到没有新消息返回，确保收集到所有数据。

修正后的代码示例 #

Consumer<String, String> consumer = consumerFactory.createConsumer();
consumer.subscribe(Collections.singletonList(topic));

// 等待分区分配完成，避免空的分区集合
Set<TopicPartition> partitions = new HashSet<>();
while (partitions.isEmpty()) {
    // 短间隔轮询，等待协调者完成分区分配
    consumer.poll(Duration.ofMillis(100));
    partitions = consumer.assignment();
}

// 将所有分区定位到最开始的偏移量
consumer.seekToBeginning(partitions);

// 循环拉取直到没有新消息，确保收集全量数据
List<ConsumerRecord<String, String>> allRecords = new ArrayList<>();
ConsumerRecords<String, String> records;
do {
    // 每次poll设置1秒超时，超时无新消息则退出循环
    records = consumer.poll(Duration.ofSeconds(1));
    records.forEach(allRecords::add);
} while (!records.isEmpty());

// 此时allRecords包含主题中的所有消息

关键配置确认 #

你已经设置的两个配置是正确的，需要确保它们生效：

• AUTO_OFFSET_RESET_CONFIG: "earliest"：确保消费者在没有偏移量记录时从头开始消费
• enable.auto.commit: false：禁止自动提交偏移量，避免下次消费时跳过消息

额外注意事项 #

• 如果主题的分区数较多，或者网络环境较差，可以适当延长分区等待的轮询时间，但短间隔轮询（如100ms）已经足够应对绝大多数场景。
• 如果消息量极大，可以调整fetch.max.records配置，增大单次poll的最大拉取条数，但对于10条消息的场景，默认值已经足够。

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