KafkaConsumer订阅主题后获取已分配分区返回空如何解决

Kafka消费者的分区分配逻辑是异步触发的：

• 调用subscribe()方法仅会标记消费者要订阅的主题，不会立刻发起和Kafka协调器的通信完成分区分配
• 分区分配动作默认只会在消费者第一次调用poll()方法拉取消息时才会触发，分配完成前assignment()方法返回的始终是空集合
• 若调用poll()时设置的超时时间过短，还未等到协调器返回分区分配结果就提前返回，也会出现assignment()返回空的情况

方案1：主动调用poll触发分配并等待结果 #

订阅主题后主动调用一次带足够超时时间的poll()，确保分区分配完成后再调用assignment()获取分区：

final KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
consumer.subscribe(java.util.List.of(topic));
// 调用带足够超时时间的poll触发分区分配
consumer.poll(Duration.ofSeconds(1));

如果需要更稳妥的校验，可以轮询等待分配结果非空：

while (consumer.assignment().isEmpty()) {
    consumer.poll(Duration.ofMillis(100));
}

方案2：使用再平衡监听器获取分配结果（推荐） #

注册ConsumerRebalanceListener监听器，在分区分配完成的回调里直接拿到分配的分区，不需要主动轮询，可靠性更高：

final KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of(topic), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 此处可以直接拿到分配的分区列表，做后续逻辑处理
        List<Integer> assignedPartitions = partitions.stream()
            .filter(tp -> tp.topic().equals(topic))
            .map(TopicPartition::partition)
            .toList();
        // 赋值给全局变量或者直接执行后续逻辑
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 分区回收前的清理逻辑，按需实现
    }
});
// 后续正常调用poll即可，分配完成会自动触发上面的onPartitionsAssigned回调

注意：你当前的getPartition()实现中直接调用get()获取分区，如果消费者被分配了多个分区，该实现只会返回匹配到的第一个分区，若业务需要处理全部分区建议调整逻辑返回列表。

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