如何检查Kafka中的某个分区是否已被分配？

可以通过Kafka Consumer API提供的assignment()方法结合集合判断来实现单个分区的检查，具体步骤如下：

1. 获取已分配的分区集合
   调用consumer.assignment()会返回当前消费者已被分配的所有TopicPartition集合，这个集合里的元素已经重写了equals()和hashCode()方法，只要topic名称和分区号一致就会被判定为相等，直接用contains()方法就能准确判断。

2. 在Stream中加入过滤逻辑
   你可以在map操作前或后加入过滤步骤，筛选出已分配的分区：

示例代码（map后过滤） #

import java.util.Set;
import java.util.stream.Collectors;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.PartitionInfo;

// 先获取已分配的分区集合
Set<TopicPartition> assignedPartitions = consumer.assignment();
List<PartitionInfo> myPartitionInfo = consumer.partitionsFor(subTopicName);

List<TopicPartition> assignedTargetPartitions = myPartitionInfo.stream()
    .map(part -> new TopicPartition(subTopicName, part.partition()))
    // 过滤出已被分配的分区
    .filter(assignedPartitions::contains)
    .collect(Collectors.toList());

示例代码（map前过滤） #

如果一定要在map操作前完成检查，可以先临时构造TopicPartition进行判断：

Set<TopicPartition> assignedPartitions = consumer.assignment();
List<PartitionInfo> myPartitionInfo = consumer.partitionsFor(subTopicName);

List<TopicPartition> assignedTargetPartitions = myPartitionInfo.stream()
    .filter(part -> {
        TopicPartition tempTp = new TopicPartition(subTopicName, part.partition());
        return assignedPartitions.contains(tempTp);
    })
    .map(part -> new TopicPartition(subTopicName, part.partition()))
    .collect(Collectors.toList());

注意事项 #

• 确保消费者已经完成分区分配：如果是刚初始化的消费者，需要先调用至少一次consumer.poll(Duration.ofMillis(100))（时长可按需调整），触发Kafka的分区分配机制，否则assignment()会返回空集合。

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