调用Consumer.Poll(Duration.ZERO)后Kafka未分配分区是什么原因

Kafka消费者的subscribe()方法仅会标记要订阅的主题，不会同步触发分区分配流程。所有和服务端协调者的交互逻辑（包括加入消费组、获取分区分配结果、发送心跳、拉取消息等），都嵌入在poll()方法的执行流程中，且和poll()调用处于同一线程。

对应现象的具体原因 #

• 你调用consumer.poll(Duration.ofMillis(0))时，超时时间为0代表poll()仅会返回当前已经缓冲好的结果，不会等待任何未完成的后台交互流程。此时分区分配请求刚触发还未拿到服务端返回的结果，consumer.assignment()返回空集合，seekToBeginning对空集合执行没有任何效果，消费者会默认从最新的位移开始消费，因此只能拿到新写入的消息。
• 调长poll()的超时时长后，poll()会在超时窗口内等待分区分配流程完成，拿到分配结果后才会继续执行后续逻辑，此时assignment()有值，seekToBeginning能正常重置所有分区的位移到最开始，因此能消费到全量历史数据。
• 线程暂停（Thread.sleep）无效的原因是：Kafka消费者的所有后台协调逻辑都依赖poll()调用驱动，线程sleep时不会执行任何网络IO和交互逻辑，分区分配流程不会有任何推进，sleep结束后assignment()依然是空集合。

和心跳机制的关联 #

心跳请求是消费者向消费组协调者证明自身存活的定时请求，它的发送同样是嵌入在poll()的执行流程中，你这个场景的核心问题不是心跳超时，而是在位移重置前没有完成分区分配流程。如果消费者长时间不调用poll()，不仅拿不到分区分配结果，还会因为没有发送心跳被协调者判定为下线，被踢出消费组。

最优修复方案 #

可以循环等待分区分配完成，避免依赖固定超时时间的不确定性：

try {
    consumer.subscribe(Collections.singletonList("names"));
    if (startingPoint != null){
        // 循环等待分区分配完成，避免超时时间设置不合理的问题
        while (consumer.assignment().isEmpty()) {
            consumer.poll(Duration.ofMillis(100));
        }
        consumer.seekToBeginning(consumer.assignment());
    }
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        keyValuePairs.add(new String[]{record.key(),record.value()});
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }

} catch (Exception e) {
    e.printStackTrace();
} finally {
    consumer.close();
}

内容的提问来源于stack exchange，提问作者Nico S.