读取Kafka Topic时如何校验无效分区编号以区分无数据和参数错误

核心思路 #

在执行分区分配、偏移量定位操作前，先主动获取目标Topic的合法分区列表，对传入的分区编号做前置校验，不合法直接抛出异常，避免无意义的消息查询操作。

实现步骤 #

• 调用KafkaConsumer.partitionsFor(String topic)方法获取目标Topic的全部分区元数据
• 提取所有合法分区编号，判断入参partition是否在合法范围内
• 校验不通过直接抛出参数异常，校验通过再执行后续的assign、seek、poll逻辑

完整代码示例 #

// 先校验分区参数合法性
List<PartitionInfo> partitionInfos = this.consumer.partitionsFor(this.topicName);
if (partitionInfos == null || partitionInfos.isEmpty()) {
    throw new IllegalArgumentException("Topic " + this.topicName + " 不存在或无法获取分区信息");
}
Set<Integer> validPartitions = partitionInfos.stream()
        .map(PartitionInfo::partition)
        .collect(Collectors.toSet());
if (!validPartitions.contains(partition)) {
    throw new IllegalArgumentException("无效分区编号: " + partition + "，Topic " + this.topicName + " 的合法分区为: " + validPartitions);
}

// 校验通过后再执行原有逻辑
TopicPartition topicPartition = new TopicPartition(this.topicName, partition);
this.consumer.assign(Collections.singletonList(topicPartition));
this.consumer.seek(topicPartition, offset);
ConsumerRecords<Object, Object> records = this.consumer.poll(50000L);

补充说明 #

partitionsFor()方法会自动同步Kafka集群的元数据，除了校验分区合法性外，还可以同时覆盖Topic不存在、Kafka集群无法连通等异常场景的校验，无需额外开发逻辑。

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