本文介绍如何在 kafka 消费者端准确判断指定 topic 分区（topicpartition）是否已被当前消费者实例分配，避免对未分配分区执行非法操作，并提供简洁可靠的代码实现与最佳实践。
本文介绍如何在 kafka 消费者端准确判断指定 topic 分区（topicpartition）是否已被当前消费者实例分配，避免对未分配分区执行非法操作，并提供简洁可靠的代码实现与最佳实践。
在 Kafka 消费者应用中，确保只对已分配给当前消费者的分区执行操作（如手动提交偏移量、初始化状态、预加载缓存等），是保障消费逻辑正确性和避免 IllegalStateException 的关键。虽然 consumer.assignment() 返回的是当前已分配的 Set
以下是一个通用、线程安全的检查方法：
public boolean isPartitionAssigned(Consumer, ?> consumer, String topic, int partition) {
TopicPartition target = new TopicPartition(topic, partition);
return consumer.assignment().contains(target);
}
该方法利用 Set.contains() 的 O(1) 平均时间复杂度，比 stream().anyMatch(...) 更简洁高效。你可将其集成到你的流式处理逻辑中：
List<partitioninfo> partitionInfos = consumer.partitionsFor(subTopicName);
List<topicpartition> assignedPartitions = partitionInfos.stream()
.map(p -> new TopicPartition(subTopicName, p.partition()))
.filter(tp -> isPartitionAssigned(consumer, tp.topic(), tp.partition()))
.collect(Collectors.toList());</topicpartition></partitioninfo>
⚠️ 重要注意事项：
- consumer.assignment() 仅反映当前已知的分配结果，它依赖于最近一次 poll() 或 subscribe() 后的再平衡完成。若在 subscribe() 后立即调用而尚未完成首次分配，该集合可能为空——此时应等待首次 poll() 触发加入组并获取分配。
- 不要依赖 partitionsFor() 的返回结果推断分配状态：该方法仅查询元数据，返回主题所有分区信息，与消费者实际分配无关。
- 在 ConsumerRebalanceListener 的 onPartitionsAssigned() 回调中，可安全假设传入的 Collection
即为最新分配；同理，在 onPartitionsRevoked() 中，这些分区已不再属于当前消费者。
✅ 推荐实践：将分区分配检查封装为工具方法，并结合 poll() 循环使用——真正的业务逻辑（如处理记录、提交偏移）天然发生在已分配分区上；若需在分配前预初始化（如构建分区级缓存），务必在 onPartitionsAssigned() 中执行，而非在 partitionsFor() 后盲目遍历。
