Kafka 集群跑久了，几乎每个运维或后端开发都会撞上同一个尴尬局面：当初建 topic 时随手给了 3 分区 2 副本，业务量一上来，消费者数量扩到 8 个，结果一半在空转；或者集群扩容加了 broker，老 topic 的副本还是死死压在原来那几台机器上，新机器闲着吃灰。手工在命令行里一个个改？几十上百个 topic 改到怀疑人生，还容易漏、容易错。这篇就聊聊怎么用代码把"批量修改 Kafka topic 分区数和副本数"这件事做扎实——不是贴一段脚本就完事，而是把背后的坑、参数怎么算、异常怎么兜底都掰开讲清楚，适合手里管着几十个 topic、又不想半夜被报警叫醒的同行参考。

1. 先搞清楚：分区和副本到底在改什么

动手写代码之前，得先把这两个概念在 Kafka 里的物理含义捋顺，否则代码写出来也是瞎改。

1.1 分区数只能加不能减的硬约束

Kafka 的分区数是一个单向操作——你可以从 3 加到 12，但永远没法从 12 减回 3。原因在于分区是消息有序性的基本单位，消息一旦按 key 哈希落到某个分区，减少分区就意味着要重新分配已有消息，这会破坏 offset 的连续性，也会让消费者组的位点彻底乱套。所以 Kafka 官方干脆把"减分区"这条路封死了。

这意味着你在写批量脚本时，必须先把"目标分区数 < 当前分区数"的 topic 过滤掉，否则脚本跑到一半抛异常，前面的改了一半后面的没改，状态就乱了。我的做法是在脚本里先做一次全量扫描，把每个 topic 的当前分区数、副本数拉出来，和目标值做对比，分成"需要扩容分区""分区已达标""分区超标需人工确认"三类，只对第一类执行操作。

1.2 副本数调整本质是分区重分配

副本数的修改和分区数不太一样。改副本数实际上是触发一次partition reassignment（分区重分配），Kafka 会把指定分区的副本在 broker 之间重新摆放。这个过程是异步的，你提交了重分配计划之后，集群会在后台慢慢搬数据，搬完之前这个 topic 处于"重分配进行中"的状态。

这里有个关键点很多人忽略：改副本数不是简单地"多复制一份"。当你把副本从 2 改到 3，Kafka 需要选一个新的 broker 来放第三个副本，同时保证同一个分区的多个副本不在同一台 broker 上（否则副本就失去意义了）。所以重分配计划必须由集群的现状推导出来，不能拍脑袋写死。

1.3 两个操作能不能一起做

可以，但要小心。分区扩容和副本调整可以合并成一次重分配提交，Kafka 的alterPartitionReassignments接口支持同时指定新的分区列表和副本分布。不过我的经验是能分开就分开：先扩分区（这个操作轻量，只是元数据变更），等新分区被消费者正常分配、流量跑稳了，再单独做副本重分配（这个涉及大量数据搬迁，IO 压力大）。两件事叠在一起做，一旦出问题很难定位是哪一步引起的。

2. 为什么不用现成命令，非要写代码

命令行工具kafka-reassign-partitions.sh和kafka-topics.sh --alter确实能改，但批量场景下它们有几个绕不过去的短板，这也是我最终选择写代码的核心原因。

2.1 命令行工具的批量能力太弱

kafka-topics.sh --alter --partitions N一次只能改一个 topic，想批量就得套 shell 循环。而副本调整更麻烦，kafka-reassign-partitions.sh要求你先手写一个 JSON 格式的重分配计划文件，格式长这样：

{ "version": 1, "partitions": [ {"topic": "order-events", "partition": 0, "replicas": [1, 2, 3]}, {"topic": "order-events", "partition": 1, "replicas": [2, 3, 1]} ] }

几十个 topic、每个几十个分区，这个 JSON 靠手写根本不现实，必须用程序生成。既然都要写程序生成计划了，那不如把整个流程都用代码串起来。

2.2 需要事务性的"预检—执行—校验"闭环

批量操作最怕的是"改了一半"。命令行工具没有预检能力，你没法在真正执行前知道哪些 topic 会失败。而代码可以先做一轮 dry-run：检查目标分区数是否合法、目标副本数是否超过 broker 数量、集群是否有 broker 处于下线状态。这些检查在代码里就是几个 if，但能挡掉 80% 的翻车场景。

2.3 需要记录和回滚依据

每次重分配前，代码可以把当前的副本分布快照存下来（写文件或存库），万一改完发现有问题，至少知道原来长什么样，能照着恢复。命令行工具做完就做完了，啥也不留。

下面这张表是我总结的命令行 vs 代码方案的对比，供你判断自己的场景该走哪条路：

3. 用 AdminClient 搭出批量修改的骨架

Java 生态里操作 Kafka 元数据的标准入口是AdminClient，它封装了 topic 管理、分区重分配、集群信息查询等一整套 API。下面按"初始化—扫描—预检—执行—校验"的顺序把骨架搭出来。

3.1 初始化 AdminClient 与集群信息拉取

第一步是把客户端建起来，同时拉一份集群的 broker 列表，后面推导副本分布全靠它。

Properties props = new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092,broker3:9092"); props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "30000"); props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "60000"); try (AdminClient admin = AdminClient.create(props)) { // 拉取集群节点信息 DescribeClusterResult cluster = admin.describeCluster(); Collection<Node> nodes = cluster.nodes().get(30, TimeUnit.SECONDS); List<Integer> brokerIds = nodes.stream() .map(Node::id) .sorted() .collect(Collectors.toList()); System.out.println("在线 broker: " + brokerIds); }

这里有个细节值得说：describeCluster().nodes()返回的是当前存活的 broker。如果集群里有 broker 临时下线，它不会出现在这个列表里。所以推导副本分布时，一定要基于这个实时列表，而不是配置文件里写死的 broker 清单，否则可能把副本分配到一台已经挂掉的机器上，重分配直接卡住。

3.2 全量扫描 topic 的当前状态

拿到 broker 列表后，把需要处理的 topic 现状摸清楚。listTopics配合describeTopics能拿到每个 topic 的分区数、每个分区的副本分布。

ListTopicsResult listResult = admin.listTopics(new ListTopicsOptions().listInternal(false)); Set<String> topicNames = listResult.names().get(30, TimeUnit.SECONDS); // 只处理匹配前缀的 topic，避免误伤内部 topic List<String> targets = topicNames.stream() .filter(name -> name.startsWith("biz-")) .collect(Collectors.toList()); DescribeTopicsResult describeResult = admin.describeTopics(targets); Map<String, TopicDescription> descriptions = describeResult.allTopicNames().get(60, TimeUnit.SECONDS); for (Map.Entry<String, TopicDescription> entry : descriptions.entrySet()) { String topic = entry.getKey(); TopicDescription desc = entry.getValue(); int partitionCount = desc.partitions().size(); // 取第一个分区的副本数作为代表（正常情况下各分区副本数一致） int replicaCount = desc.partitions().get(0).replicas().size(); System.out.printf("topic=%s, partitions=%d, replicas=%d%n", topic, partitionCount, replicaCount); }

注意：listInternal(false)这个参数别漏，否则__consumer_offsets这类内部 topic 也会被扫进来，一旦被误改，消费者位点全乱，后果很严重。

3.3 预检：把不合法的目标挡在执行之前

扫描完之后，进入预检环节。这一步是纯逻辑判断，但价值极高。我一般会检查这几项：

• 目标分区数是否小于当前值：小于则跳过并记录，因为 Kafka 不支持减分区。
• 目标副本数是否大于在线 broker 数：大于则无法分配，直接报错。
• 目标副本数是否小于 1：非法值，拦截。
• topic 是否正处于重分配中：通过listPartitionReassignments查询，正在搬数据的 topic 先跳过，避免叠加操作。

// 查询正在重分配的 topic ListPartitionReassignmentsResult reassignResult = admin.listPartitionReassignments(); Set<String> reassigningTopics = reassignResult.reassignments().get(30, TimeUnit.SECONDS) .keySet().stream() .map(TopicPartitionReplica::topic) .collect(Collectors.toSet()); for (String topic : targets) { if (reassigningTopics.contains(topic)) { System.out.println("跳过（重分配进行中）: " + topic); continue; } // 其余校验逻辑... }

预检结果我习惯输出成一张表，人工过一眼再执行，尤其是生产环境，多这一步能省掉很多事后解释。

4. 副本重分配计划怎么推导才靠谱

这是整个批量脚本里技术含量最高的部分。分区扩容简单，调个接口传个数字就行；副本重分配则要自己算出"每个分区的每个副本该放在哪台 broker 上"。

4.1 轮询分配的基本思路

最朴素也最常用的策略是轮询（round-robin）：把 broker 列表排好序，对每个分区的第 i 个副本，选择brokerList[(partitionId + i) % brokerCount]作为落点。这样能保证：

• 同一分区的多个副本落在不同 broker 上（因为 i 不同，取模结果不同，前提是副本数 ≤ broker 数）；
• 不同分区的副本分布错开，负载相对均衡。

List<Integer> brokers = brokerIds; // 已排序 int replicaCount = 3; List<TopicPartitionReplica> assignments = new ArrayList<>(); for (int partition = 0; partition < partitionCount; partition++) { List<Integer> replicas = new ArrayList<>(); for (int i = 0; i < replicaCount; i++) { int brokerId = brokers.get((partition + i) % brokers.size()); replicas.add(brokerId); } assignments.add(new TopicPartitionReplica(topic, partition, replicas)); }

4.2 为什么不能无脑轮询：机架感知的考量

轮询在 broker 数量少、机架信息不敏感的场景下够用。但如果集群跨机架部署，无脑轮询可能把同一分区的三个副本全塞进同一个机架，一旦那个机架断电，分区直接不可用。Kafka 本身支持机架感知（broker.rack配置），但重分配计划是你自己生成的，Kafka 不会帮你纠正。

所以更稳妥的做法是：如果集群配置了机架信息，先按机架分组，保证每个分区的副本尽量分散到不同机架。实现上可以先按机架把 broker 分桶，然后每个分区的副本从不同桶里轮流取。这块逻辑稍复杂，但如果你的集群确实跨机房，值得花时间做。

4.3 保留原有副本的"最小搬迁"策略

轮询有个副作用：它会把所有分区的副本分布全部重排，哪怕原来某些副本的位置是合理的。这会导致不必要的数据搬迁——本来只需要加一个副本，结果三个副本全换了位置，搬了 3 倍的数据。

更聪明的做法是"最小搬迁"：对每个分区，先保留原有的副本（只要它们还在线），只对缺失的副本位置做补充。比如原来分区 0 的副本在 [1, 2]，目标副本数是 3，那就保留 1 和 2，从剩下的 broker 里挑一个（比如 3）补上，最终变成 [1, 2, 3]。这样只搬一个新副本，IO 压力小得多。

// 最小搬迁：保留原有在线副本，补充新副本 List<Integer> currentReplicas = desc.partitions().get(partition).replicas(); List<Integer> newReplicas = new ArrayList<>(currentReplicas); for (int brokerId : brokers) { if (newReplicas.size() >= replicaCount) break; if (!newReplicas.contains(brokerId)) { newReplicas.add(brokerId); } }

这个策略在"只加副本不换机器"的场景下几乎是零搬迁，实测下来对线上流量影响最小。

4.4 提交重分配与进度轮询

计划生成好之后，通过alterPartitionReassignments提交。提交是异步的，返回后需要轮询listPartitionReassignments看进度。

Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>(); for (TopicPartitionReplica r : assignments) { reassignments.put( new TopicPartition(r.topic(), r.partition()), Optional.of(new NewPartitionReassignment(r.replicas())) ); } AlterPartitionReassignmentsResult alterResult = admin.alterPartitionReassignments(reassignments); alterResult.all().get(60, TimeUnit.SECONDS); // 轮询进度 while (true) { Map<TopicPartition, PartitionReassignment> ongoing = admin.listPartitionReassignments().reassignments().get(30, TimeUnit.SECONDS); if (ongoing.isEmpty()) { System.out.println("重分配完成"); break; } System.out.println("进行中分区数: " + ongoing.size()); Thread.sleep(5000); }

轮询间隔别设太短，5 到 10 秒比较合适。太频繁会给 controller 增加压力，太稀疏又感知不到卡死。

5. 分区扩容的独立处理与顺序编排

副本重分配讲完了，回头说分区扩容。它比副本调整简单，但顺序编排上有讲究。

5.1 扩容分区的接口调用

分区扩容用createPartitions接口，传入 topic 名和目标总数（注意是总数，不是增量）。

Map<String, NewPartitions> newPartitions = new HashMap<>(); newPartitions.put("biz-order-events", NewPartitions.increaseTo(12)); CreatePartitionsResult result = admin.createPartitions(newPartitions); result.all().get(30, TimeUnit.SECONDS);

increaseTo(12)的意思是"把分区数增加到 12"，如果当前是 8，就新增 4 个。如果当前已经是 12 或更多，接口会抛InvalidPartitionsException，所以预检那一步的过滤是必须的。

5.2 为什么先扩分区再调副本

前面提过要分开做，这里说清楚原因。分区扩容后，新分区需要被消费者组重新分配（触发 rebalance），同时新分区的 leader 选举、ISR 建立都需要时间。如果这时候立刻做副本重分配，两件事的元数据变更会交织在一起，controller 的压力陡增，而且一旦出问题，你分不清是扩容引起的还是重分配引起的。

我的标准流程是：扩分区 → 等消费者 rebalance 完成、新分区流量正常 → 再做副本重分配。中间这个"等"没有固定时长，看业务流量，一般观察 10 到 30 分钟。

5.3 分区数怎么定才不浪费

分区数不是越多越好。每个分区在 broker 上都对应文件句柄和内存开销，分区太多会拖慢 controller 的元数据处理。经验公式是：分区数 ≈ 期望吞吐 / 单分区吞吐，同时考虑消费者并行度——分区数最好是你最大消费者实例数的整数倍，这样每个消费者分到的分区数均匀。

举个实际例子：某业务峰值 50MB/s，单分区实测能扛 10MB/s，那至少 5 个分区；消费者最多部署 8 个实例，为了让 8 个实例都能分到活，分区数取 8 的倍数，最终定 8 或 16。别一上来就拍个 100 分区，那是给自己挖坑。

6. 踩过的坑：这些异常不处理会很难受

代码骨架搭起来能跑，但生产环境跑批量操作，异常处理才是分水岭。下面几个是我实际踩过、或者见别人踩过的。

6.1 重分配卡在"进行中"不动

最常见的是重分配提交后，进度一直卡着。原因通常是目标 broker 磁盘满了或者网络带宽打满，数据搬不过去。代码里如果只是死等，就会一直挂着。我的处理是给轮询加超时，超过阈值（比如 30 分钟没进展）就告警并记录当前状态，人工介入。

还有一种情况是 broker 在重分配过程中下线，Kafka 会暂停搬迁，等 broker 回来。这种在日志里能看到ReplicaNotAvailableException，代码里捕获后应该标记该 topic 为"待重试"，而不是直接失败退出。

6.2 副本数超过 broker 数的静默失败

如果目标副本数设成 5，但集群只有 3 台 broker，alterPartitionReassignments会抛InvalidReplicaAssignmentException。这个异常在批量循环里如果不捕获，整个批次就中断了。所以每个 topic 的操作要包在独立的 try-catch 里，单个失败不影响其他 topic，最后汇总报告。

6.3 内部 topic 被误伤

前面提过listInternal(false)，但还有一种隐蔽情况：某些业务 topic 名字恰好以__开头，被误判为内部 topic 跳过，或者反过来。稳妥的做法是维护一个显式的白名单或前缀规则，别依赖自动判断。

6.4 权限不足导致的半途而废

如果执行脚本的账号只有部分 topic 的 ALTER 权限，跑到没权限的 topic 时会抛ClusterAuthorizationException。批量场景下，建议先做一轮权限预检：对每个目标 topic 尝试describeTopics，能描述的基本就有改的权限（具体看 ACL 配置），把没权限的提前剔除。

下面这张表汇总了常见异常和应对方式：

7. 上线前的自测清单与灰度节奏

批量改 Kafka 元数据，最忌讳直接在生产全量跑。我一般按这个节奏来。

7.1 先在测试集群跑通全流程

测试集群不需要和生产一样大，3 台 broker 就够验证逻辑。重点验证：分区扩容后消费者能否正常 rebalance、副本重分配后 ISR 是否正常、脚本的预检和异常分支是否按预期走。测试集群上可以故意造几个异常场景（比如把目标副本数设成 10），看脚本是不是优雅地报错而不是崩掉。

7.2 生产环境先灰度一个 topic

挑一个流量小、非核心的 topic 先跑一遍，观察重分配进度、集群 IO、消费者延迟。确认没问题后，再按批次扩大范围，每批之间留出观察窗口。我习惯每批不超过 10 个 topic，批间间隔 15 分钟。

7.3 执行前必备的快照

每次执行前，把当前所有目标 topic 的分区数、副本分布导出成 JSON 存好。万一改完发现某个 topic 的副本分布不理想，至少能照着快照重新生成一份"恢复计划"。这个习惯救过我一次——某次轮询策略把副本全排到了两台机器上，靠快照快速回滚了。

# 快照示例（伪代码，实际用 AdminClient 导出） kafka-topics.sh --describe --topic biz-order-events > snapshot_$(date +%s).txt

8. 几个让脚本更耐用的工程化细节

最后聊几个提升脚本健壮性的点，都是实际用下来觉得值的。

8.1 幂等性设计

脚本要能重复跑而不产生副作用。判断依据是"当前状态是否已等于目标状态"——分区数已达标的跳过扩容，副本分布已符合目标的跳过重分配。这样即使中途失败，重跑一遍也能接着往下走，不用手工清理。

8.2 并发度控制

批量操作不要开太多并发。重分配本身就会占用大量网络和磁盘 IO，如果同时对几十个 topic 提交重分配，集群直接被打爆。我的做法是串行提交、异步等待：一次只提交一个 topic 的重分配，等它完成再提交下一个。慢是慢点，但稳。

8.3 日志要能定位到具体 topic 和分区

批量脚本的日志如果只打"操作失败"，排查起来很痛苦。每条日志至少带上 topic 名、分区号、操作类型、异常摘要。我一般用结构化日志，方便后续 grep 或导入分析。

8.4 把配置外置

目标分区数、目标副本数、topic 前缀、broker 列表这些别写死在代码里，抽到配置文件或命令行参数。不同环境（测试、预发、生产）用不同配置，代码一份就够。

# config.yaml 示例 target: partitionCount: 12 replicaCount: 3 filter: topicPrefix: "biz-" excludeTopics: - "biz-temp-debug"

这套东西搭下来，一个几十 topic 的集群，批量调整从原来手工折腾大半天，变成跑个脚本喝杯咖啡的功夫。真正花时间的不是写代码，而是把预检、异常、快照这些"防御性"的部分做扎实——毕竟改 Kafka 元数据这事，稳比快重要得多。我个人在实际操作中的体会是，凡是涉及副本重分配的批量操作，宁可多花十分钟做预检和灰度，也别图快一把梭，事后救火的成本远高于事前多等的那些时间。