单消费者分配多分区时Kafka consumer seek()致poll()数据丢失问题

1. 该现象是否为预期行为？ #

是预期行为，核心原因在于当前代码逻辑的缺陷，结合Kafka消费者的位置（position）管理机制共同导致：

• Kafka消费者每次调用poll()后，会自动将各分区的消费位置更新为本次拉取到的最大偏移量+1，后续poll()会从这个位置开始拉取新数据，除非手动调用seek()调整。
• 你的代码在单条记录处理失败后，仅对当前失败记录所在分区执行seek()，然后直接break循环，导致同一次poll()返回的其他分区未处理记录被直接跳过。这些未处理记录所在的分区，消费位置已经被自动推进，且你没有对这些分区执行seek()回退偏移量，后续poll()会从推进后的位置开始拉取，完全跳过这些未处理的记录，最终造成数据丢失。
• 多消费者各分配一个分区时，每个消费者只处理单个分区，break不会影响其他分区的处理，因此不会出现丢失问题。

2. 可用的Kafka API解决方案及优化思路 #

针对单消费者多分区的场景，需要调整处理逻辑，确保每个分区的偏移量能正确管理，避免跳过未处理记录，以下是可行的方案：

方案1：按分区分组处理，确保分区内顺序重试 #

将poll()拉取的记录按分区分组，逐个分区处理。每个分区内的记录顺序处理，若某条记录失败，仅对当前分区执行seek()，并终止当前分区的处理，后续循环仅处理其他正常分区，下一次poll()会重新拉取该失败分区的未处理记录。

改进后的示例代码：

ConsumerRecords<String, String> consumerRecords = consumer.poll(100);
if (!consumerRecords.isEmpty()) {
    // 按分区分组记录
    Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsByPartition = new HashMap<>();
    for (ConsumerRecord<String, String> record : consumerRecords) {
        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
        recordsByPartition.computeIfAbsent(tp, k -> new ArrayList<>()).add(record);
    }

    for (Map.Entry<TopicPartition, List<ConsumerRecord<String, String>>> entry : recordsByPartition.entrySet()) {
        TopicPartition tp = entry.getKey();
        List<ConsumerRecord<String, String>> partitionRecords = entry.getValue();
        boolean partitionProcessFailed = false;
        long lastSuccessOffset = -1;

        for (ConsumerRecord<String, String> record : partitionRecords) {
            if (!ProcessinApplication(record.topic(), record.value(), record.partition())) {
                // 处理失败，seek到当前记录偏移量，标记分区处理失败
                consumer.seek(tp, record.offset());
                partitionProcessFailed = true;
                break;
            }
            lastSuccessOffset = record.offset();
        }

        // 只有分区内所有记录处理成功，才提交该分区的偏移量（注意偏移量要+1，指向下次拉取的位置）
        if (!partitionProcessFailed && lastSuccessOffset != -1) {
            try {
                Map<TopicPartition, OffsetAndMetadata> partitionAndOffset = new HashMap<>();
                partitionAndOffset.put(tp, new OffsetAndMetadata(lastSuccessOffset + 1));
                consumer.commitSync(partitionAndOffset);
            } catch (Exception e) {
                // 处理提交异常
                e.printStackTrace();
            }
        }
    }
}

方案2：结合pause()和resume()控制分区消费 #

当某个分区处理失败时，调用consumer.pause(tp)暂停该分区的消费，避免后续poll()拉取该分区的新数据，专注重试失败的记录；当重试成功后，再调用consumer.resume(tp)恢复该分区的正常消费。

方案3：改用批量偏移量提交 #

放弃逐条提交偏移量，改为每个分区处理完本次poll()拉取的所有记录后，再提交该分区的最大偏移量。若处理过程中出现失败，直接seek()到该分区本次拉取的起始偏移量，下一次poll()会重新拉取整个批次的记录，确保不丢失。

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