1. 为什么“批量改分区副本数”是个绕不开的活儿干过几年消息中间件运维的人都有一个共同体会集群刚上线那会儿topic 的分区数和副本数往往是拍脑袋定的业务量小的时候一切正常等到流量涨上来、机器扩了几轮、机房调整过几次之后原来那套配置就开始处处别扭。分区数不够消费端吞吐上不去堆积告警天天响副本数不合理要么写入延迟高得离谱要么某台 broker 一挂就出现欠副本运维半夜被叫起来处理。这时候你面对的不是一个 topic而是几十上百个 topic手动一个个改改到怀疑人生还容易漏、容易错。这篇内容聊的就是批量修改 Kafka topic 分区数与副本数的代码实现思路和落地细节。核心解决三个问题一是把“逐个手改”变成“脚本批量跑”二是把“改多少、怎么改”变成可计算、可校验的参数三是把“改完有没有风险”变成可观测、可回滚的流程。适合谁看适合已经能熟练使用 Kafka 基础命令、日常负责集群运维或平台开发的工程师也适合刚接手消息队列、想搞清楚分区和副本到底怎么调的新手——文中会把关键原理和参数计算过程都摊开讲不假设你什么都懂。需要先明确一个前提Kafka 的分区数只能增不能减副本数可以增减但涉及数据复制两者都不是“改个配置重启”那么轻量。批量操作的本质是把单次操作的风险乘以 N所以代码写出来只是第一步真正值钱的是围绕代码的那套判断逻辑和防护措施。下面从设计思路开始一层层拆。2. 整体方案设计与技术选型考量2.1 三种实现路径的取舍批量修改 topic 配置市面上常见三条路纯命令行脚本、AdminClient 编程、以及基于管理平台的 API 调用。我先把三者的特点摆出来再说我为什么最终选了 AdminClient 为主、脚本为辅的组合。实现方式优点缺点适用场景命令行脚本kafka-topics.sh 循环上手快无需写代码输出解析脆弱错误处理弱难以做前置校验topic 数量少、一次性操作AdminClient 编程类型安全可做校验、重试、并发控制需要写代码要处理版本兼容批量、可重复、需审计管理平台 API有界面操作留痕依赖平台能力灵活性受限有成熟平台的团队命令行脚本的问题在于“脆”。kafka-topics.sh --describe的输出是给人看的文本你用 awk、grep 去解析分区数和副本数一旦 Kafka 版本升级改了输出格式脚本就悄悄失效更可怕的是失效时可能解析出错误的值导致后续计算出错。而 AdminClient 拿到的是结构化对象TopicDescription里分区、副本、ISR 都是明确的字段判断逻辑可以写得很扎实。我最终的做法是用 AdminClient 写一个可配置的批处理程序命令行脚本只作为应急兜底。程序读取一份“目标配置清单”对每个 topic 做前置检查、计算目标值、执行修改、校验结果、输出报告。这样整个流程可重复执行跑第二遍时已经达标的 topic 会被自动跳过天然具备幂等性。2.2 分区数与副本数的调整边界写代码之前必须把规则吃透否则代码逻辑再漂亮也是错的。这里有两个硬约束分区数只能增加不能减少。Kafka 的分区是消息有序性和并行度的基本单位减少分区意味着要把多个分区的数据合并消息顺序无法保证所以官方直接不支持。代码里如果检测到目标分区数小于当前值必须报错跳过而不是“尽力而为”。副本数可以增减但减少副本有数据丢失风险窗口。把副本从 3 降到 2需要下线一个副本如果此时恰好有 broker 故障可能触发欠副本甚至不可用。所以减副本操作要格外谨慎通常只在缩容场景下做且要确认集群健康。还有一个容易被忽略的点分区数增加后基于 key 的消息分区归属会变化。原来 key 为 “order-123” 的消息固定落在分区 2分区数从 6 扩到 12 之后同样的 key 可能落到分区 8。如果下游消费逻辑依赖“同一 key 在同一分区”来做状态聚合扩分区就会打破这个假设。这一点代码没法自动判断只能靠人工在配置清单里标注哪些 topic 不能随便扩程序读取标注后跳过。2.3 配置清单驱动的设计我采用“声明式配置 程序执行”的模式。运维人员不直接操作集群而是维护一份清单写明每个 topic 的目标分区数和副本数程序负责把现实状态向目标状态收敛。清单大概长这样用 YAML 便于人工维护topics: - name: order-events partitions: 12 replicationFactor: 3 allowPartitionIncrease: true - name: user-profile partitions: 6 replicationFactor: 2 allowPartitionIncrease: false # 依赖 key 分区禁止扩容这种设计的好处是变更意图被显式记录可以进代码仓库做版本管理谁在什么时候把哪个 topic 改成多少一目了然程序每次运行都是“对账”而不是“盲改”天然支持重复执行和灰度。相比“写死一堆 topic 名的脚本”可维护性高出一个量级。3. 核心代码实现与关键细节拆解3.1 环境准备与依赖引入代码基于 Java 实现用 Maven 管理依赖。AdminClient 从 Kafka 0.11 开始提供建议客户端版本与集群大版本对齐避免 API 不兼容。核心依赖就一个dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version3.6.0/version /dependency版本选择上有个经验客户端版本不要低于集群版本太多也不要高于太多。低太多可能缺少新 API比如增量修改副本的alterPartitionReassignments在老版本没有高太多可能用到集群不支持的协议特性。稳妥做法是客户端版本与集群主版本保持一致或略低一个小版本。创建 AdminClient 的配置项里bootstrap.servers是必须的另外建议显式设置request.timeout.ms和default.api.timeout.ms。批量操作时单次请求可能较慢默认超时偏短容易误报失败。我的习惯是Properties props new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker1:9092,broker2:9092); props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000); props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 60000); props.put(AdminClientConfig.RETRIES_CONFIG, 3);注意retries对幂等操作友好但副本重分配这类操作本身有状态重试前要先查当前状态不能无脑重试后文会讲。3.2 读取现状把集群真实状态摸清楚第一步永远是“读现状”。用admin.describeTopics()拿到所有目标 topic 的描述从中提取当前分区数、每个分区的副本列表、ISR 列表。这里有个细节describeTopics返回的TopicDescription里partitions()给出每个分区的PartitionInforeplicas()是副本所在 broker 列表inSyncReplicas()是当前同步的副本。判断集群健康的关键指标是 ISR 数量是否等于副本数如果某个分区 ISR 少于副本数说明有副本落后此时做副本调整风险很高应该先跳过并告警。MapString, TopicDescription descMap admin.describeTopics(topicNames).all().get(); for (Map.EntryString, TopicDescription e : descMap.entrySet()) { TopicDescription td e.getValue(); int currentPartitions td.partitions().size(); int currentRf td.partitions().get(0).replicas().size(); boolean healthy td.partitions().stream() .allMatch(p - p.inSyncReplicas().size() p.replicas().size()); // 记录 currentPartitions、currentRf、healthy 供后续决策 }这里假设同一 topic 各分区副本数一致——绝大多数情况如此但严格来说 Kafka 允许分区级副本数不同虽然不推荐。代码里取第一个分区的副本数作为代表同时可以加一个校验如果发现各分区副本数不一致标记为“异常 topic”不自动处理交给人工确认。这种“宁可少动不可乱动”的保守策略在批量运维里是保命原则。3.3 分区扩容的计算与执行分区扩容用admin.createPartitions()传入 topic 名和目标总分区数注意是总数不是增量。执行前要做三重校验目标值大于当前值、topic 标记允许扩容、集群健康。计算上没什么复杂公式但有个实践中的建议值单 broker 承载的分区总数要控制。经验上单 broker 分区数控制在 2000 到 4000 以内比较稳具体取决于堆内存和文件句柄数。所以扩容前可以算一下扩容后总分区数 ÷ broker 数如果超过阈值就告警。NewPartitions newPartitions NewPartitions.increaseTo(targetPartitions); MapString, NewPartitions request Collections.singletonMap(topicName, newPartitions); CreatePartitionsResult result admin.createPartitions(request); result.all().get();createPartitions是异步的.get()会阻塞等待完成。批量场景下不建议一个 topic 一个 topic 串行.get()可以先把所有请求发出去再统一收集结果但要注意并发度控制——同时改太多 topic 会给 controller 造成压力。我的做法是用固定大小的线程池并发度设为 3 到 5实测比较稳。实操心得分区扩容后新分区不会自动均衡已有数据只是新消息会按规则写入新分区。如果业务对数据分布敏感扩容后可以配合分区重分配工具做一次均衡但那是另一个话题且重分配本身开销大要挑低峰期做。3.4 副本数调整增量重分配才是正解副本数调整比分区扩容复杂得多因为它涉及数据在 broker 之间的搬移。老版本用kafka-reassign-partitions.sh生成重分配方案再执行新版本 AdminClient 提供了alterPartitionReassignments可以增量地指定每个分区的新副本集合。核心思路是对每个分区构造一个“目标副本列表”长度等于目标副本数然后提交重分配。目标副本怎么选不能随便挑 broker要考虑机架感知如果配置了broker.rack和负载均衡。简化版做法是从当前副本列表里保留前 N 个N 为目标副本数如果当前副本数少于目标则从集群可用 broker 里补充。补充时优先选当前承载分区数较少的 broker避免热点。// 以单分区为例构造目标副本列表 ListInteger currentReplicas partitionInfo.replicas().stream() .map(Node::id).collect(Collectors.toList()); ListInteger targetReplicas new ArrayList(currentReplicas.subList(0, Math.min(currentReplicas.size(), targetRf))); while (targetReplicas.size() targetRf) { int candidate pickLeastLoadedBroker(clusterBrokers, targetReplicas); targetReplicas.add(candidate); }构造好所有分区的目标副本后组装成MapTopicPartition, OptionalNewPartitionReassignment提交MapTopicPartition, OptionalNewPartitionReassignment reassignments new HashMap(); for (int i 0; i targetReplicasPerPartition.size(); i) { TopicPartition tp new TopicPartition(topicName, i); reassignments.put(tp, Optional.of(new NewPartitionReassignment(targetReplicasPerPartition.get(i)))); } admin.alterPartitionReassignments(reassignments).all().get();这里有个大坑重分配是异步且耗时的提交成功不代表完成。数据量大的 topic 可能搬几个小时。代码提交后必须轮询listPartitionReassignments()查看进度直到所有分区不再出现在进行中的列表里。轮询间隔建议 10 到 30 秒太频繁会给集群增加无谓压力。3.5 结果校验与幂等保障每轮操作结束后重新describeTopics一次比对实际值与目标值。全部达标才算成功否则输出差异清单。这一步是幂等性的关键程序下次运行时达标的 topic 直接跳过只处理有差异的所以同一份清单可以反复跑不会产生副作用。校验逻辑要覆盖三个维度分区数是否等于目标、副本数是否等于目标、ISR 是否全部同步。第三个维度容易被忽略——副本数改对了但 ISR 没跟上说明有副本在同步中或已落后此时集群并未真正健康。我的报告里会把“配置达标但 ISR 未满”单独列出来提示观察一段时间再确认。4. 实操全流程与现场记录4.1 一次完整的批量调整过程拿一个模拟场景走一遍。假设集群有 6 台 broker清单里有 20 个 topic其中 15 个需要扩分区8 个需要调副本有重叠。流程分五步预检阶段读取清单逐个 topic 拉取现状校验目标值合法性分区只增不减、副本数在 1 到 broker 数之间、集群健康度ISR 是否满。预检不通过的 topic 直接进“跳过清单”附原因。分区扩容阶段对需要扩分区的 topic 提交createPartitions并发度 3每个请求等待完成。记录每个 topic 的耗时。副本调整阶段对需要调副本的 topic 构造重分配方案提交alterPartitionReassignments然后进入轮询。这一步耗时最长实测一个 50GB 的 topic 从 2 副本扩到 3 副本在千兆网络下大约 8 到 15 分钟。校验阶段全部操作完成后统一 describe 一次生成达标/未达标报告。输出报告把预检跳过、执行成功、执行失败、校验未通过四类结果分别列出附上每个 topic 的前后对比。现场记录里有个值得说的细节第 3 步轮询时我发现某个 topic 的重分配卡在 90% 很久不动。查listPartitionReassignments发现是某个分区的目标副本所在 broker 磁盘 IO 打满同步速度极慢。处理办法是等而不是取消——中途取消重分配会留下中间状态反而更麻烦。这也印证了前面说的副本调整要挑低峰期做且提前确认目标 broker 的磁盘和网络余量。4.2 参数计算的实际例子举个具体的数某 topic 当前 6 分区、2 副本日均消息量 2 亿条单条平均 1KB峰值吞吐约 50MB/s。业务反馈消费端有堆积想扩到 12 分区、3 副本。分区数计算消费端并行度受分区数限制假设单个消费者实例处理能力约 8MB/s50MB/s 峰值需要至少 7 个消费者取整到 12 分区留足余量。副本数从 2 到 3是为了容忍单 broker 故障而不影响写入2 副本时挂一台就只剩 1 副本写入可能受影响。存储估算12 分区 × 3 副本 × 单分区日均数据量。单分区日均约 2 亿 ÷ 12 ≈ 1667 万条 × 1KB ≈ 16GB乘 3 副本约 48GB再乘保留天数假设 3 天约 144GB。确认集群剩余空间足够后再执行。这个计算过程看着简单但实际运维中因为漏算副本倍数导致磁盘写满的事故并不少见。4.3 并发度与超时的调优记录第一版代码我用了 10 的并发度结果在 20 个 topic 的批量里controller 出现了明显的请求排队部分createPartitions超时。把并发降到 3 之后全部请求在 2 分钟内完成无超时。结论是分区扩容的瓶颈在 controller 的元数据写入不在网络并发度不宜高。副本重分配的瓶颈在数据搬移并发度可以稍高5 左右但同样受限于目标 broker 的磁盘带宽。超时设置上createPartitions单次 30 秒足够alterPartitionReassignments提交本身很快但后续轮询的总时长要设一个上限比如 2 小时超过就判定为“超时未完成”输出当前进度供人工介入而不是无限等下去。5. 常见问题与排查技巧实录5.1 高频问题速查表现象可能原因排查与处理createPartitions 报 InvalidPartitionsException目标分区数小于当前值检查清单分区只能增不能减重分配提交后长时间无进展目标 broker 磁盘/网络瓶颈查 broker 磁盘 IO 和网络等待或换目标 broker副本数改完但 ISR 不满副本同步中或 broker 负载高观察一段时间持续不满则查 broker 日志部分 topic 被跳过预检不通过ISR 不满/标记禁止扩容看报告里的跳过原因人工确认后重跑客户端报 UnsupportedVersionException客户端与集群版本不匹配对齐客户端版本或改用兼容 API5.2 几个踩过的坑坑一以为副本数改完就万事大吉。有次批量把一批 topic 从 2 副本改成 3 副本提交后看alterPartitionReassignments返回成功就收工了。第二天发现其中几个 topic 的 ISR 一直是 2查下来是新增的副本所在 broker 磁盘快满了同步一直失败。教训是提交成功 ≠ 完成必须轮询到重分配列表清空再校验 ISR。坑二忽略机架感知导致跨机架流量暴涨。集群配了broker.rack但我的目标副本选择逻辑只按负载挑 broker没考虑机架结果把三个副本全放到了同一机架既失去了容灾意义又让跨机架同步流量分布不均。后来在选副本时加了机架去重逻辑优先保证副本分布在不同机架。坑三清单里 topic 名写错程序静默跳过。早期版本对“清单里有但集群里不存在”的 topic 只是打个日志就跳过结果有人把order-events写成order-event程序没报错运维以为改完了。后来改成清单里的 topic 在集群中不存在时直接判定为配置错误并中断执行强制人工修正。独家技巧批量操作前先用--dry-run模式跑一遍。程序只做预检和计算不提交任何变更输出“将要做什么”的完整计划。确认计划无误后再去掉 dry-run 真正执行。这个习惯帮我拦下过好几次清单配置错误。5.3 回滚与应急分区扩容不可逆所以没有“回滚”一说只能靠事前谨慎。副本调整理论上可以改回去但同样要经历一次数据搬移成本不低。所以我的原则是任何批量变更前先对目标 topic 做一次元数据快照把 describe 结果存成文件万一出问题至少知道变更前的状态是什么便于评估影响和制定恢复方案。快照文件按时间戳命名保留最近若干份。应急场景下如果发现某个 topic 改完导致消费异常第一反应不是改回去而是先确认异常原因——是分区扩容导致 key 分布变化还是副本调整期间短暂的不可用。前者需要业务侧适配后者通常等重分配完成就恢复。盲目回滚可能引发二次抖动。6. 把批处理做成可持续的工具代码跑通只是起点。真正让这套东西产生长期价值的是把它固化成一个团队可用的工具清单进版本库、执行有 dry-run、结果有报告、异常有告警。我在实际项目里把这套逻辑封装成了一个内部小工具运维同学改完清单提个合并请求评审通过后由流水线触发执行执行报告自动归档。这样一来“批量改分区副本”从一件需要小心翼翼的手工活变成了一个有记录、可追溯、可复现的常规操作。最后分享一个我在实际使用中的体会批量操作的代码写起来花的时间往往不到整体工作量的三成剩下七成都在前置校验、并发控制、结果校验和异常处理上。新手容易把精力全放在“怎么调 API”上而有经验的人会把大部分心思花在“怎么保证不出事”上。分区和副本这两个参数改对了是优化改错了就是事故代码只是执行手段判断力才是核心。清单里那几行allowPartitionIncrease的标注、报告里那几列 ISR 状态看着不起眼关键时刻能救命。 
长河君 · 2026/10/26 12:40:31 · 阅读 1.2k Kafka批量修改分区与副本数：AdminClient实现与运维实践
1. 为什么“批量改分区副本数”是个绕不开的活儿干过几年消息中间件运维的人都有一个共同体会集群刚上线那会儿topic 的分区数和副本数往往是拍脑袋定的业务量小的时候一切正常等到流量涨上来、机器扩了几轮、机房调整过几次之后原来那套配置就开始处处别扭。分区数不够消费端吞吐上不去堆积告警天天响副本数不合理要么写入延迟高得离谱要么某台 broker 一挂就出现欠副本运维半夜被叫起来处理。这时候你面对的不是一个 topic而是几十上百个 topic手动一个个改改到怀疑人生还容易漏、容易错。这篇内容聊的就是批量修改 Kafka topic 分区数与副本数的代码实现思路和落地细节。核心解决三个问题一是把“逐个手改”变成“脚本批量跑”二是把“改多少、怎么改”变成可计算、可校验的参数三是把“改完有没有风险”变成可观测、可回滚的流程。适合谁看适合已经能熟练使用 Kafka 基础命令、日常负责集群运维或平台开发的工程师也适合刚接手消息队列、想搞清楚分区和副本到底怎么调的新手——文中会把关键原理和参数计算过程都摊开讲不假设你什么都懂。需要先明确一个前提Kafka 的分区数只能增不能减副本数可以增减但涉及数据复制两者都不是“改个配置重启”那么轻量。批量操作的本质是把单次操作的风险乘以 N所以代码写出来只是第一步真正值钱的是围绕代码的那套判断逻辑和防护措施。下面从设计思路开始一层层拆。2. 整体方案设计与技术选型考量2.1 三种实现路径的取舍批量修改 topic 配置市面上常见三条路纯命令行脚本、AdminClient 编程、以及基于管理平台的 API 调用。我先把三者的特点摆出来再说我为什么最终选了 AdminClient 为主、脚本为辅的组合。实现方式优点缺点适用场景命令行脚本kafka-topics.sh 循环上手快无需写代码输出解析脆弱错误处理弱难以做前置校验topic 数量少、一次性操作AdminClient 编程类型安全可做校验、重试、并发控制需要写代码要处理版本兼容批量、可重复、需审计管理平台 API有界面操作留痕依赖平台能力灵活性受限有成熟平台的团队命令行脚本的问题在于“脆”。kafka-topics.sh --describe的输出是给人看的文本你用 awk、grep 去解析分区数和副本数一旦 Kafka 版本升级改了输出格式脚本就悄悄失效更可怕的是失效时可能解析出错误的值导致后续计算出错。而 AdminClient 拿到的是结构化对象TopicDescription里分区、副本、ISR 都是明确的字段判断逻辑可以写得很扎实。我最终的做法是用 AdminClient 写一个可配置的批处理程序命令行脚本只作为应急兜底。程序读取一份“目标配置清单”对每个 topic 做前置检查、计算目标值、执行修改、校验结果、输出报告。这样整个流程可重复执行跑第二遍时已经达标的 topic 会被自动跳过天然具备幂等性。2.2 分区数与副本数的调整边界写代码之前必须把规则吃透否则代码逻辑再漂亮也是错的。这里有两个硬约束分区数只能增加不能减少。Kafka 的分区是消息有序性和并行度的基本单位减少分区意味着要把多个分区的数据合并消息顺序无法保证所以官方直接不支持。代码里如果检测到目标分区数小于当前值必须报错跳过而不是“尽力而为”。副本数可以增减但减少副本有数据丢失风险窗口。把副本从 3 降到 2需要下线一个副本如果此时恰好有 broker 故障可能触发欠副本甚至不可用。所以减副本操作要格外谨慎通常只在缩容场景下做且要确认集群健康。还有一个容易被忽略的点分区数增加后基于 key 的消息分区归属会变化。原来 key 为 “order-123” 的消息固定落在分区 2分区数从 6 扩到 12 之后同样的 key 可能落到分区 8。如果下游消费逻辑依赖“同一 key 在同一分区”来做状态聚合扩分区就会打破这个假设。这一点代码没法自动判断只能靠人工在配置清单里标注哪些 topic 不能随便扩程序读取标注后跳过。2.3 配置清单驱动的设计我采用“声明式配置 程序执行”的模式。运维人员不直接操作集群而是维护一份清单写明每个 topic 的目标分区数和副本数程序负责把现实状态向目标状态收敛。清单大概长这样用 YAML 便于人工维护topics: - name: order-events partitions: 12 replicationFactor: 3 allowPartitionIncrease: true - name: user-profile partitions: 6 replicationFactor: 2 allowPartitionIncrease: false # 依赖 key 分区禁止扩容这种设计的好处是变更意图被显式记录可以进代码仓库做版本管理谁在什么时候把哪个 topic 改成多少一目了然程序每次运行都是“对账”而不是“盲改”天然支持重复执行和灰度。相比“写死一堆 topic 名的脚本”可维护性高出一个量级。3. 核心代码实现与关键细节拆解3.1 环境准备与依赖引入代码基于 Java 实现用 Maven 管理依赖。AdminClient 从 Kafka 0.11 开始提供建议客户端版本与集群大版本对齐避免 API 不兼容。核心依赖就一个dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version3.6.0/version /dependency版本选择上有个经验客户端版本不要低于集群版本太多也不要高于太多。低太多可能缺少新 API比如增量修改副本的alterPartitionReassignments在老版本没有高太多可能用到集群不支持的协议特性。稳妥做法是客户端版本与集群主版本保持一致或略低一个小版本。创建 AdminClient 的配置项里bootstrap.servers是必须的另外建议显式设置request.timeout.ms和default.api.timeout.ms。批量操作时单次请求可能较慢默认超时偏短容易误报失败。我的习惯是Properties props new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker1:9092,broker2:9092); props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000); props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 60000); props.put(AdminClientConfig.RETRIES_CONFIG, 3);注意retries对幂等操作友好但副本重分配这类操作本身有状态重试前要先查当前状态不能无脑重试后文会讲。3.2 读取现状把集群真实状态摸清楚第一步永远是“读现状”。用admin.describeTopics()拿到所有目标 topic 的描述从中提取当前分区数、每个分区的副本列表、ISR 列表。这里有个细节describeTopics返回的TopicDescription里partitions()给出每个分区的PartitionInforeplicas()是副本所在 broker 列表inSyncReplicas()是当前同步的副本。判断集群健康的关键指标是 ISR 数量是否等于副本数如果某个分区 ISR 少于副本数说明有副本落后此时做副本调整风险很高应该先跳过并告警。MapString, TopicDescription descMap admin.describeTopics(topicNames).all().get(); for (Map.EntryString, TopicDescription e : descMap.entrySet()) { TopicDescription td e.getValue(); int currentPartitions td.partitions().size(); int currentRf td.partitions().get(0).replicas().size(); boolean healthy td.partitions().stream() .allMatch(p - p.inSyncReplicas().size() p.replicas().size()); // 记录 currentPartitions、currentRf、healthy 供后续决策 }这里假设同一 topic 各分区副本数一致——绝大多数情况如此但严格来说 Kafka 允许分区级副本数不同虽然不推荐。代码里取第一个分区的副本数作为代表同时可以加一个校验如果发现各分区副本数不一致标记为“异常 topic”不自动处理交给人工确认。这种“宁可少动不可乱动”的保守策略在批量运维里是保命原则。3.3 分区扩容的计算与执行分区扩容用admin.createPartitions()传入 topic 名和目标总分区数注意是总数不是增量。执行前要做三重校验目标值大于当前值、topic 标记允许扩容、集群健康。计算上没什么复杂公式但有个实践中的建议值单 broker 承载的分区总数要控制。经验上单 broker 分区数控制在 2000 到 4000 以内比较稳具体取决于堆内存和文件句柄数。所以扩容前可以算一下扩容后总分区数 ÷ broker 数如果超过阈值就告警。NewPartitions newPartitions NewPartitions.increaseTo(targetPartitions); MapString, NewPartitions request Collections.singletonMap(topicName, newPartitions); CreatePartitionsResult result admin.createPartitions(request); result.all().get();createPartitions是异步的.get()会阻塞等待完成。批量场景下不建议一个 topic 一个 topic 串行.get()可以先把所有请求发出去再统一收集结果但要注意并发度控制——同时改太多 topic 会给 controller 造成压力。我的做法是用固定大小的线程池并发度设为 3 到 5实测比较稳。实操心得分区扩容后新分区不会自动均衡已有数据只是新消息会按规则写入新分区。如果业务对数据分布敏感扩容后可以配合分区重分配工具做一次均衡但那是另一个话题且重分配本身开销大要挑低峰期做。3.4 副本数调整增量重分配才是正解副本数调整比分区扩容复杂得多因为它涉及数据在 broker 之间的搬移。老版本用kafka-reassign-partitions.sh生成重分配方案再执行新版本 AdminClient 提供了alterPartitionReassignments可以增量地指定每个分区的新副本集合。核心思路是对每个分区构造一个“目标副本列表”长度等于目标副本数然后提交重分配。目标副本怎么选不能随便挑 broker要考虑机架感知如果配置了broker.rack和负载均衡。简化版做法是从当前副本列表里保留前 N 个N 为目标副本数如果当前副本数少于目标则从集群可用 broker 里补充。补充时优先选当前承载分区数较少的 broker避免热点。// 以单分区为例构造目标副本列表 ListInteger currentReplicas partitionInfo.replicas().stream() .map(Node::id).collect(Collectors.toList()); ListInteger targetReplicas new ArrayList(currentReplicas.subList(0, Math.min(currentReplicas.size(), targetRf))); while (targetReplicas.size() targetRf) { int candidate pickLeastLoadedBroker(clusterBrokers, targetReplicas); targetReplicas.add(candidate); }构造好所有分区的目标副本后组装成MapTopicPartition, OptionalNewPartitionReassignment提交MapTopicPartition, OptionalNewPartitionReassignment reassignments new HashMap(); for (int i 0; i targetReplicasPerPartition.size(); i) { TopicPartition tp new TopicPartition(topicName, i); reassignments.put(tp, Optional.of(new NewPartitionReassignment(targetReplicasPerPartition.get(i)))); } admin.alterPartitionReassignments(reassignments).all().get();这里有个大坑重分配是异步且耗时的提交成功不代表完成。数据量大的 topic 可能搬几个小时。代码提交后必须轮询listPartitionReassignments()查看进度直到所有分区不再出现在进行中的列表里。轮询间隔建议 10 到 30 秒太频繁会给集群增加无谓压力。3.5 结果校验与幂等保障每轮操作结束后重新describeTopics一次比对实际值与目标值。全部达标才算成功否则输出差异清单。这一步是幂等性的关键程序下次运行时达标的 topic 直接跳过只处理有差异的所以同一份清单可以反复跑不会产生副作用。校验逻辑要覆盖三个维度分区数是否等于目标、副本数是否等于目标、ISR 是否全部同步。第三个维度容易被忽略——副本数改对了但 ISR 没跟上说明有副本在同步中或已落后此时集群并未真正健康。我的报告里会把“配置达标但 ISR 未满”单独列出来提示观察一段时间再确认。4. 实操全流程与现场记录4.1 一次完整的批量调整过程拿一个模拟场景走一遍。假设集群有 6 台 broker清单里有 20 个 topic其中 15 个需要扩分区8 个需要调副本有重叠。流程分五步预检阶段读取清单逐个 topic 拉取现状校验目标值合法性分区只增不减、副本数在 1 到 broker 数之间、集群健康度ISR 是否满。预检不通过的 topic 直接进“跳过清单”附原因。分区扩容阶段对需要扩分区的 topic 提交createPartitions并发度 3每个请求等待完成。记录每个 topic 的耗时。副本调整阶段对需要调副本的 topic 构造重分配方案提交alterPartitionReassignments然后进入轮询。这一步耗时最长实测一个 50GB 的 topic 从 2 副本扩到 3 副本在千兆网络下大约 8 到 15 分钟。校验阶段全部操作完成后统一 describe 一次生成达标/未达标报告。输出报告把预检跳过、执行成功、执行失败、校验未通过四类结果分别列出附上每个 topic 的前后对比。现场记录里有个值得说的细节第 3 步轮询时我发现某个 topic 的重分配卡在 90% 很久不动。查listPartitionReassignments发现是某个分区的目标副本所在 broker 磁盘 IO 打满同步速度极慢。处理办法是等而不是取消——中途取消重分配会留下中间状态反而更麻烦。这也印证了前面说的副本调整要挑低峰期做且提前确认目标 broker 的磁盘和网络余量。4.2 参数计算的实际例子举个具体的数某 topic 当前 6 分区、2 副本日均消息量 2 亿条单条平均 1KB峰值吞吐约 50MB/s。业务反馈消费端有堆积想扩到 12 分区、3 副本。分区数计算消费端并行度受分区数限制假设单个消费者实例处理能力约 8MB/s50MB/s 峰值需要至少 7 个消费者取整到 12 分区留足余量。副本数从 2 到 3是为了容忍单 broker 故障而不影响写入2 副本时挂一台就只剩 1 副本写入可能受影响。存储估算12 分区 × 3 副本 × 单分区日均数据量。单分区日均约 2 亿 ÷ 12 ≈ 1667 万条 × 1KB ≈ 16GB乘 3 副本约 48GB再乘保留天数假设 3 天约 144GB。确认集群剩余空间足够后再执行。这个计算过程看着简单但实际运维中因为漏算副本倍数导致磁盘写满的事故并不少见。4.3 并发度与超时的调优记录第一版代码我用了 10 的并发度结果在 20 个 topic 的批量里controller 出现了明显的请求排队部分createPartitions超时。把并发降到 3 之后全部请求在 2 分钟内完成无超时。结论是分区扩容的瓶颈在 controller 的元数据写入不在网络并发度不宜高。副本重分配的瓶颈在数据搬移并发度可以稍高5 左右但同样受限于目标 broker 的磁盘带宽。超时设置上createPartitions单次 30 秒足够alterPartitionReassignments提交本身很快但后续轮询的总时长要设一个上限比如 2 小时超过就判定为“超时未完成”输出当前进度供人工介入而不是无限等下去。5. 常见问题与排查技巧实录5.1 高频问题速查表现象可能原因排查与处理createPartitions 报 InvalidPartitionsException目标分区数小于当前值检查清单分区只能增不能减重分配提交后长时间无进展目标 broker 磁盘/网络瓶颈查 broker 磁盘 IO 和网络等待或换目标 broker副本数改完但 ISR 不满副本同步中或 broker 负载高观察一段时间持续不满则查 broker 日志部分 topic 被跳过预检不通过ISR 不满/标记禁止扩容看报告里的跳过原因人工确认后重跑客户端报 UnsupportedVersionException客户端与集群版本不匹配对齐客户端版本或改用兼容 API5.2 几个踩过的坑坑一以为副本数改完就万事大吉。有次批量把一批 topic 从 2 副本改成 3 副本提交后看alterPartitionReassignments返回成功就收工了。第二天发现其中几个 topic 的 ISR 一直是 2查下来是新增的副本所在 broker 磁盘快满了同步一直失败。教训是提交成功 ≠ 完成必须轮询到重分配列表清空再校验 ISR。坑二忽略机架感知导致跨机架流量暴涨。集群配了broker.rack但我的目标副本选择逻辑只按负载挑 broker没考虑机架结果把三个副本全放到了同一机架既失去了容灾意义又让跨机架同步流量分布不均。后来在选副本时加了机架去重逻辑优先保证副本分布在不同机架。坑三清单里 topic 名写错程序静默跳过。早期版本对“清单里有但集群里不存在”的 topic 只是打个日志就跳过结果有人把order-events写成order-event程序没报错运维以为改完了。后来改成清单里的 topic 在集群中不存在时直接判定为配置错误并中断执行强制人工修正。独家技巧批量操作前先用--dry-run模式跑一遍。程序只做预检和计算不提交任何变更输出“将要做什么”的完整计划。确认计划无误后再去掉 dry-run 真正执行。这个习惯帮我拦下过好几次清单配置错误。5.3 回滚与应急分区扩容不可逆所以没有“回滚”一说只能靠事前谨慎。副本调整理论上可以改回去但同样要经历一次数据搬移成本不低。所以我的原则是任何批量变更前先对目标 topic 做一次元数据快照把 describe 结果存成文件万一出问题至少知道变更前的状态是什么便于评估影响和制定恢复方案。快照文件按时间戳命名保留最近若干份。应急场景下如果发现某个 topic 改完导致消费异常第一反应不是改回去而是先确认异常原因——是分区扩容导致 key 分布变化还是副本调整期间短暂的不可用。前者需要业务侧适配后者通常等重分配完成就恢复。盲目回滚可能引发二次抖动。6. 把批处理做成可持续的工具代码跑通只是起点。真正让这套东西产生长期价值的是把它固化成一个团队可用的工具清单进版本库、执行有 dry-run、结果有报告、异常有告警。我在实际项目里把这套逻辑封装成了一个内部小工具运维同学改完清单提个合并请求评审通过后由流水线触发执行执行报告自动归档。这样一来“批量改分区副本”从一件需要小心翼翼的手工活变成了一个有记录、可追溯、可复现的常规操作。最后分享一个我在实际使用中的体会批量操作的代码写起来花的时间往往不到整体工作量的三成剩下七成都在前置校验、并发控制、结果校验和异常处理上。新手容易把精力全放在“怎么调 API”上而有经验的人会把大部分心思花在“怎么保证不出事”上。分区和副本这两个参数改对了是优化改错了就是事故代码只是执行手段判断力才是核心。清单里那几行allowPartitionIncrease的标注、报告里那几列 ISR 状态看着不起眼关键时刻能救命。