如何使用Java API修改Kafka Topic的副本因子？

完全可以通过Java的AdminClient实现，官方API本身就支持修改已存在Topic的副本分配（间接修改副本因子），具体步骤和代码示例如下：

核心思路 #

修改副本因子本质是重新分配Topic各分区的副本列表，通过AdminClient.alterReplicasAssignments()方法完成，步骤分为：获取当前分区信息、构造新的副本分配方案、执行修改、验证结果。

代码示例 #

1. 初始化AdminClient并执行修改 #

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterReplicasAssignmentsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.ReplicaAssignment;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartition;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaTopicReplicaUpdater {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 配置AdminClient连接信息
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092");
        try (AdminClient adminClient = AdminClient.create(props)) {
            String targetTopic = "your-topic-name";
            // 获取当前Topic的分区详情
            DescribeTopicsResult describeResult = adminClient.describeTopics(Collections.singletonList(targetTopic));
            TopicDescription topicDesc = describeResult.all().get().get(targetTopic);

            // 构造新的副本分配方案（示例：将副本因子从1改为3，分配到broker 0、1、2）
            Map<TopicPartition, Optional<ReplicaAssignment>> newAssignments = new HashMap<>();
            for (var partitionInfo : topicDesc.partitions()) {
                int partitionId = partitionInfo.partition();
                // 新副本列表的长度即为目标副本因子
                var newReplicas = Arrays.asList(0, 1, 2);
                // ReplicaAssignment参数：副本列表、同步副本列表（通常与副本列表一致）
                ReplicaAssignment assignment = new ReplicaAssignment(newReplicas, newReplicas);
                newAssignments.put(new TopicPartition(targetTopic, partitionId), Optional.of(assignment));
            }

            // 执行副本分配修改
            AlterReplicasAssignmentsResult alterResult = adminClient.alterReplicasAssignments(newAssignments);
            alterResult.all().get(); // 阻塞等待操作完成

            // 验证修改结果
            TopicDescription updatedDesc = adminClient.describeTopics(Collections.singletonList(targetTopic)).all().get().get(targetTopic);
            updatedDesc.partitions().forEach(partition -> 
                System.out.printf("分区%d的副本列表：%s%n", partition.partition(), partition.replicas())
            );
        }
    }
}

注意事项 #

• 新的副本因子不能超过集群中可用的Broker数量，否则会抛出IllegalArgumentException。
• 执行操作的账号需要具备Topic的ALTER权限，否则会返回权限错误。
• 修改过程中Kafka会自动在新副本上同步数据，无需手动干预，大Topic可能需要较长时间完成同步。
• 如果是减少副本因子，需确保移除的副本不是当前分区的leader（Kafka会自动选举新leader，但建议提前确认分区leader分布）。

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