Kafka Java Consumer无法读取活跃段记录的问题排查

问题背景 #

我正在测试Kafka的日志压缩功能，使用kafka-clients 3.3.1版本，创建了配置如下的压缩主题：

cleanup.policy=compact
delete.retention.ms=100
segment.ms=100
min.cleanable.dirty.ratio=0.01

同时将broker的log.retention.check.interval.ms设为1秒，确保每秒执行日志压缩。

按顺序发送3批消息：

#1
"k1":"V1"
"k3":"V1"
"k2":"V1"
"k1":"V2"
#2 
"k2":"V2"
"k1":"V3"
"k1":"V4"
#3
"k4":"V1"

每批间隔至少3秒等待压缩完成。控制台消费者能正常获取到所有最终消息：

"k3":"V1"
"k2":"V2"
"k1":"V4"
"k4":"V1"

但使用Java KafkaConsumer时，k4消息丢失，仅返回：

"k3":"V1"
"k2":"V2"
"k1":"V4"

通过kafka-run-class.sh kafka.tools.DumpLogSegments工具确认k4存在于head segment（00000000000000000007.log）中。

问题原因 #

你的Java消费者代码仅执行了单次poll操作，且超时时间仅为1秒：

final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(1));

Kafka消费者启动后，需要完成分区分配、元数据同步等初始化操作，单次短超时的poll可能无法覆盖到head segment中的新消息。而控制台消费者是持续循环poll的，因此能获取到所有消息。

另外，生产者仅调用了flush()，未等待消息完全提交到broker的确认，可能存在消息还未完全落盘就启动消费者的情况。

解决方案 #

• 循环执行poll操作：将消费者改为循环poll，直到获取到所有预期消息或超时：

@SneakyThrows
public static void main(String[] args) {
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getConsumerProperties());) {
        consumer.subscribe(Collections.singletonList(TOPIC_NAME));
        // 循环poll，持续5秒或获取到k4为止
        long endTime = System.currentTimeMillis() + 5000;
        boolean foundK4 = false;
        while (System.currentTimeMillis() < endTime && !foundK4) {
            final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> consumerRecord : poll) {
                log.info("Key : {} with value : {}", consumerRecord.key(), consumerRecord.value());
                if ("k4".equals(consumerRecord.key())) {
                    foundK4 = true;
                }
            }
        }
    }
}

• 确保生产者消息完全提交：在生产者发送消息时，等待发送确认，而不只是flush：

// 替换原send代码，等待消息确认
producer.send(new ProducerRecord<>(Constants.TOPIC_NAME, "k4", "V1")).get();
producer.flush();

• 适当延长poll超时时间：如果不需要循环，可适当延长单次poll的超时时间，确保消费者有足够时间完成初始化并拉取所有消息：

final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(5));

补充说明 #

Kafka的head segment是当前活跃的写入段，不会被压缩，但消费者完全可以读取其中的内容。问题并非消费者排除了head segment，而是单次短时间的poll未能拉取到该段的消息。

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