Kafka至少一次语义下重复消息场景及poll()重复返回疑问

测试环境 #

• 单主题：foo
• 主题foo仅包含1个分区
• 单个消费者，group.id为test

示例代码（手动偏移量控制） #

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test");
props.setProperty("max.poll.records", "50"); // 每次拉取最多50条消息
props.setProperty("auto.offset.reset", "earliest"); // 无提交偏移量时从头开始消费
props.setProperty("enable.auto.commit", "false"); // 关闭自动提交，手动控制偏移量
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("foo")); // 订阅目标主题
final int minBatchSize = 200;
List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        buffer.add(record);
    }
    if (buffer.size() >= minBatchSize) {
        insertIntoDb(buffer); // 批量写入数据库
        consumer.commitSync(); // 同步提交偏移量
        buffer.clear();
    }
}

已知重复消息场景 #

• 应用崩溃未提交偏移量：前4次poll()拉取共200条消息，成功写入数据库，但提交偏移量前应用崩溃。重启后，消费者因无有效提交记录，按照auto.offset.reset=earliest从头消费，导致200条消息重复入库。
• 消费者重平衡未提交偏移量：同一消费组内存在2个消费者，其中一个在提交偏移量前崩溃触发重平衡，新接管分区的消费者会从上次成功提交的偏移量开始消费（若未提交则从头），导致已处理的消息重复。

核心逻辑：消费者始终从最后一次成功提交的偏移量开始消费，若消息处理完成但未提交偏移量就发生故障，消息会被重新处理，这是至少一次语义的核心表现。

疑问解答 #

1. 是否存在其他导致重复消息的场景？ #

存在以下常见场景：

• 提交偏移量失败但业务已完成：消息处理完成后，调用commitSync()时因网络波动、Broker故障等原因提交失败，应用重启后会从上次成功提交的偏移量重新消费，导致已处理消息重复。
• 手动提交错误偏移量：代码逻辑错误，提交了比实际已处理位置更小的偏移量（比如仅提交了批次中前半部分消息的偏移量），后续poll()会拉取已处理过的消息。
• 消费者偏移量元数据丢失：Kafka存储消费者偏移量的__consumer_offsets主题数据丢失，消费者重启后会触发auto.offset.reset策略，从头消费导致重复。

2. 缓冲区是否可能包含重复消息？ #

正常健康的消费流程中，代码里的buffer不会自动出现重复消息，除非：

• poll()返回了重复消息（正常场景下不会发生）；
• 代码逻辑错误（比如重复执行buffer.add(record)操作）；
• 消费回溯后，重新拉取的已处理消息被再次加入缓冲区（比如重启后从头消费的场景）。

3. 无异常、无崩溃的正常情况下，连续两次poll()是否可能返回重复消息？ #

不会。单个健康消费者在无手动修改偏移量（如seek()）、无重平衡、无提交失败回溯的情况下，内部会严格跟踪当前消费的位置，每次poll()都会从下一个未消费的偏移量开始拉取消息，返回的批次不会出现重叠。

4. 单个健康消费者内存中的offset.position是否始终保持顺序性？是否poll()出现问题必抛异常，否则返回无重叠的消息批次？ #

• 单个健康消费者的offset.position始终保持严格递增的顺序，不会出现倒退（除非手动调用seek()修改位置）。
• 若poll()过程中出现问题（如网络故障、Broker不可用、超时等），会直接抛出对应的异常（如TimeoutException、InterruptException等）；只要poll()成功返回，消息批次一定是无重叠、按偏移量递增的。

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