如何在Spring Boot中为Kafka Topic动态添加分区？

Kafka完全支持在运行时为已存在的Topic添加分区（仅支持增加，不能减少），下面是补全你现有代码后的实现：

public void addPartitionIfNotExists(int partitionId){
    Map<String, TopicDescription> games = kafkaAdmin.describeTopics(Collections.singleton("games"));
    TopicDescription gamesTopicDescription = games.get("games");
    List<TopicPartitionInfo> partitionsInfo = gamesTopicDescription.partitions();
    boolean partitionIdExists = partitionsInfo.stream().anyMatch(partitionInfo -> partitionInfo.partition() == partitionId);
    
    if (!partitionIdExists){
        // 计算需要新增的分区数：目标分区ID+1减去现有分区总数（分区ID从0开始计数）
        int currentPartitionCount = partitionsInfo.size();
        int partitionsToAdd = (partitionId + 1) - currentPartitionCount;
        
        if (partitionsToAdd > 0) {
            // 构建新增分区配置，指定最终分区总数
            NewPartitions newPartitions = NewPartitions.increaseTo(currentPartitionCount + partitionsToAdd);
            // 执行添加分区操作，同步等待结果完成
            kafkaAdmin.createPartitions(Collections.singletonMap("games", newPartitions)).all().get();
        }
    }
}

关键说明 #

• 调用createPartitions后，通过all().get()同步等待操作完成，也可改用异步回调处理结果
• 新增分区会触发消费者组重新平衡，需确保消费者能正常处理新分区的消息
• 若Topic使用自定义分区器，要确认新分区能被正确分配消息，避免数据倾斜或路由异常
• 执行该操作的客户端需拥有Kafka集群的ALTER权限

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