未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用边界与维护说明

本文面向具备 Kafka 架构与运维基础的读者，用于在完成环境适配的测试集群中学习扩容、分区重分配、Leader 平衡、滚动重启及权限配额管理，不是可直接复制执行的生产手册。新增 Broker 配置和 AclAuthorizer 示例以 ZooKeeper 模式为主，不能直接套用于 KRaft 或任意版本升级；ACL 示例未提供所依赖的 admin-client.properties 与服务端 JAAS 或等效认证配置，滚动重启脚本依赖自行生成和维护的 /var/run/kafka.pid，迁移后的 --throttle -1 --execute 清理方式尚需结合具体工具版本及实际残留配置确认。生产操作须评估副本、ISR、客户端容错和升级路径，迁移与重启均不保证业务零影响。本次仅作全文结构审阅，未作外部事实核验或命令执行验证。已取消无法确定目标的可点击引用「02-Kafka架构与核心概念」，保留原提及；Missing link target: 02-Kafka架构与核心概念

Kafka 集群管理与运维 - 扩容、重分配与日常操作手册

abstract 本节概览

学习目标：掌握 Kafka 集群扩容、分区重分配、Topic 管理、消费者组管理以及权限配额管理
前置知识：性能调优实战、Kafka 架构
预计用时：35 分钟
后续学习：监控与故障排查

集群扩容流程

集群扩容分为两个阶段：新增 Broker 节点 和 分区数据迁移到新节点。新 Broker 加入后不会自动承接已有分区，必须通过分区重分配来实现负载均衡。

第一步：准备新 Broker 节点

# 1. 在新节点安装 Kafka，配置与现有节点一致
# 2. 编辑 server.properties，关键参数：

# 新 Broker 的 ID，必须在集群中唯一（不能与现有 Broker ID 重复）
broker.id=4

# ZooKeeper 连接地址（与现有集群一致）
zookeeper.connect=zk-01:2181,zk-02:2181,zk-03:2181/kafka

# 数据目录（与现有节点规范一致）
log.dirs=/data1/kafka,/data2/kafka

# 监听地址
listeners=PLAINTEXT://kafka-04:9092
advertised.listeners=PLAINTEXT://kafka-04:9092

# 3. 启动新 Broker
bin/kafka-server-start.sh -daemon config/server.properties

# 4. 验证新 Broker 已加入集群（在任意现有节点执行）
# 列出所有 Broker，确认新 broker.id=4 出现
bin/kafka-broker-api-versions.sh --bootstrap-server kafka-01:9092 | grep "kafka-04"

# 或通过 ZooKeeper 确认
bin/zookeeper-shell.sh zk-01:2181/kafka ls /brokers/ids

note Kafka 3.x KRaft 模式

Kafka 3.x 引入了 KRaft 模式（无 ZooKeeper），新增 Broker 时需要通过 kafka-storage.sh 和 Controller Quorum 完成节点注册，具体参考官方 KRaft 文档。

第二步：生成分区重分配方案

新 Broker 上线后，已有 Topic 的分区仍在旧 Broker 上，需要生成重分配方案将部分分区迁移过来。

# 创建待重分配的 Topic 列表文件
cat > topics-to-move.json << 'EOF'
{
  "topics": [
    {"topic": "orders"},
    {"topic": "payments"},
    {"topic": "user-events"}
  ],
  "version": 1
}
EOF

# 生成重分配方案
# --broker-list：包含所有 Broker ID（含新加入的 4）
# 工具会自动计算将分区均匀分布到所有 Broker 的方案
bin/kafka-reassign-partitions.sh \
    --bootstrap-server kafka-01:9092 \
    --topics-to-move-json-file topics-to-move.json \
    --broker-list "1,2,3,4" \
    --generate

输出包含两部分：当前分配方案（Current partition replica assignment）和建议的新方案（Proposed partition reassignment configuration）。将新方案保存为文件：

# 保存建议的重分配方案
cat > reassignment-plan.json << 'EOF'
{
  "version": 1,
  "partitions": [
    {"topic": "orders", "partition": 0, "replicas": [1, 2, 4]},
    {"topic": "orders", "partition": 1, "replicas": [2, 3, 4]},
    {"topic": "orders", "partition": 2, "replicas": [3, 4, 1]},
    {"topic": "payments", "partition": 0, "replicas": [4, 1, 2]}
  ]
}
EOF

第三步：执行重分配

# 执行重分配（限制迁移带宽，避免影响线上业务）
bin/kafka-reassign-partitions.sh \
    --bootstrap-server kafka-01:9092 \
    --reassignment-json-file reassignment-plan.json \
    --throttle 50000000 \
    --execute

tip throttle 参数

--throttle 50000000 将迁移带宽限制为 50 MB/s。生产环境建议在业务低峰期执行，并根据网络带宽调整此值。迁移完成后记得取消限速：

bin/kafka-reassign-partitions.sh \
    --bootstrap-server kafka-01:9092 \
    --reassignment-json-file reassignment-plan.json \
    --throttle -1 \
    --execute

第四步：验证重分配进度

# 查看重分配状态
bin/kafka-reassign-partitions.sh \
    --bootstrap-server kafka-01:9092 \
    --reassignment-json-file reassignment-plan.json \
    --verify

输出各分区的状态：

Reassignment of partition orders-0 is still in progress —— 迁移中
Reassignment of partition orders-0 is complete —— 已完成
Reassignment of partition orders-0 failed —— 失败，需排查

等待所有分区均显示 complete 后，扩容完成。此时通过 kafka-topics.sh --describe 确认各 Broker 上的分区分布是否均匀。

分区副本重分配

除了扩容场景，以下情况也需要进行分区重分配：

某台 Broker 磁盘快满，需要将部分分区迁移到其他 Broker
集群缩容前，需要将目标 Broker 上的分区先迁走
手动调整副本因子

手动构造重分配方案

{
  "version": 1,
  "partitions": [
    {
      "topic": "my-topic",
      "partition": 0,
      "replicas": [2, 3, 1],
      "log_dirs": ["any", "any", "any"]
    },
    {
      "topic": "my-topic",
      "partition": 1,
      "replicas": [3, 1, 2],
      "log_dirs": ["any", "any", "any"]
    }
  ]
}

replicas 数组的第一个元素是 Preferred Replica，即首选 Leader 副本。重分配完成不保证 Leader 立即切换到该副本；可在其处于 ISR 等条件满足时，通过 Preferred Leader 选举或自动 Leader 平衡进行调整。log_dirs 可以指定 "any" 让 Kafka 自动选择，或指定具体路径实现跨磁盘迁移。

Leader 平衡

集群运行一段时间后，由于 Broker 重启等原因，某些分区的 Leader 可能不在 Preferred Replica 上，导致 Broker 间负载不均。

手动触发 Leader 平衡

# 对所有 Topic 执行 Preferred Leader 选举
bin/kafka-leader-election.sh \
    --bootstrap-server kafka-01:9092 \
    --election-type PREFERRED \
    --all-topic-partitions

# 对特定 Topic 执行
bin/kafka-leader-election.sh \
    --bootstrap-server kafka-01:9092 \
    --election-type PREFERRED \
    --topic orders \
    --partition 0

自动 Leader 平衡配置

# server.properties
# 开启自动 Leader 平衡（默认 true）
auto.leader.rebalance.enable=true

# 不均衡阈值：某 Broker 的首选副本中，未担任 Leader 的分区比例超过 10% 时触发平衡检查
leader.imbalance.per.broker.percentage=10

# 检查间隔（默认 5 分钟）
leader.imbalance.check.interval.seconds=300

note 关于 Preferred Leader

Preferred Leader 是副本列表（replicas 数组）中的第一个副本。分区重分配时，将负载轻的 Broker 放到副本列表第一位，可以通过 Leader 平衡实现 Broker 间的读写均衡。

滚动重启

滚动重启（Rolling Restart）是逐台重启 Broker、尽量降低业务影响的标准运维操作，用于上线配置变更、兼容的版本升级或故障后恢复。能否保持业务可用取决于副本和 ISR 状态、min.insync.replicas、客户端重试能力及具体升级路径，不能保证零中断。

滚动重启步骤

# 对每台 Broker 重复以下步骤，顺序执行，不可并行

# 步骤 1：确认副本健康，并评估客户端连接及重试影响
# Kafka 客户端通常按元数据直连 Broker；不要将关闭 advertised.listeners 当作通用动态摘流手段
# 如部署确实使用负载均衡器，应按实际连接拓扑评估摘流措施

# 步骤 2：优雅停止 Broker
# 优雅停止会将 Leadership 主动迁移给其他副本，减少 ISR 收缩
bin/kafka-server-stop.sh
# 或发送 SIGTERM 信号
kill -SIGTERM $(cat /var/run/kafka.pid)

# 步骤 3：等待 Broker 进程退出
while kill -0 $(cat /var/run/kafka.pid) 2>/dev/null; do
    echo "Waiting for Kafka to stop..."
    sleep 5
done
echo "Kafka stopped."

# 步骤 4：修改配置或升级版本（按需）

# 步骤 5：启动 Broker
bin/kafka-server-start.sh -daemon config/server.properties

# 步骤 6：等待该 Broker 上的分区同步完成（ISR 恢复）
# 关键：Under-Replicated Partitions 必须归零后，才能重启下一台
bin/kafka-topics.sh \
    --bootstrap-server kafka-01:9092 \
    --describe \
    --under-replicated-partitions

warning 滚动重启的关键等待条件

每台 Broker 重启后，必须等待 Under-Replicated Partitions = 0（无欠复制分区）再重启下一台。否则可能出现某分区所有副本都不在 ISR 中，导致数据不可用。

可用以下命令持续监控，直到输出为空：

watch -n 5 "bin/kafka-topics.sh --bootstrap-server kafka-01:9092 \
  --describe --under-replicated-partitions"

Topic 管理

创建 Topic

# 创建 Topic（12 个分区，3 副本）
bin/kafka-topics.sh \
    --bootstrap-server kafka-01:9092 \
    --create \
    --topic my-topic \
    --partitions 12 \
    --replication-factor 3 \
    --config retention.ms=604800000 \
    --config compression.type=lz4

# 查看 Topic 详情
bin/kafka-topics.sh \
    --bootstrap-server kafka-01:9092 \
    --describe \
    --topic my-topic

修改 Topic 配置

# 修改消息保留时间（覆盖 Broker 默认值）
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type topics \
    --entity-name my-topic \
    --alter \
    --add-config retention.ms=259200000

# 删除自定义配置，恢复为 Broker 默认值
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type topics \
    --entity-name my-topic \
    --alter \
    --delete-config retention.ms

# 查看 Topic 当前配置
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type topics \
    --entity-name my-topic \
    --describe

增加分区数

# 将分区数从 12 增加到 24
# 注意：分区数只能增加，不能减少！
bin/kafka-topics.sh \
    --bootstrap-server kafka-01:9092 \
    --alter \
    --topic my-topic \
    --partitions 24

warning 增加分区数的注意事项

分区数只能增加，不能减少（减少需要删除并重建 Topic）
增加分区后，已有消息不会重新分布，新消息才会写入新分区
如果消息的 Key 相同的记录需要进入同一分区（有序性保证），增加分区数会破坏这一保证，需要谨慎评估

删除 Topic

# 删除 Topic（要求 delete.topic.enable=true，Kafka 1.0 后默认开启）
bin/kafka-topics.sh \
    --bootstrap-server kafka-01:9092 \
    --delete \
    --topic my-topic

# 确认删除状态（删除是异步的，短时间内 Topic 可能仍然显示）
bin/kafka-topics.sh \
    --bootstrap-server kafka-01:9092 \
    --list

消费者组管理

查看消费者组

# 列出所有消费者组
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --list

# 查看某个消费者组的详细信息（含 Lag）
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --group my-consumer-group \
    --describe

--describe 输出解读：

GROUP           TOPIC     PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID  HOST  CLIENT-ID
my-group        orders    0          1234567         1234600         33   consumer-1   host1  app-1
my-group        orders    1          9876543         9876543         0    consumer-2   host2  app-2

字段	说明
CURRENT-OFFSET	消费者当前已提交的偏移量
LOG-END-OFFSET	分区日志末端偏移量，即下一条待追加消息的位置，而非最后一条消息的偏移量
LAG	LOG-END-OFFSET - CURRENT-OFFSET，即消息积压量
CONSUMER-ID	消费者实例标识，为空表示该分区当前无消费者（组已停止）

重置消费者偏移量

# 重置到最新（跳过所有已有消息，从现在开始消费）
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --group my-consumer-group \
    --topic orders \
    --reset-offsets \
    --to-latest \
    --execute

# 重置到最早（重新消费所有历史消息）
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --group my-consumer-group \
    --topic orders \
    --reset-offsets \
    --to-earliest \
    --execute

# 重置到指定时间点（用于数据回溯）
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --group my-consumer-group \
    --topic orders \
    --reset-offsets \
    --to-datetime 2024-03-01T00:00:00.000 \
    --execute

# 重置到指定偏移量
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --group my-consumer-group \
    --topic orders:0 \
    --reset-offsets \
    --to-offset 1000000 \
    --execute

warning 重置偏移量前必须停止消费者

偏移量重置要求消费者组中所有消费者实例都已停止。如果消费者仍在运行，--execute 会报错。可以先用 --dry-run 替代 --execute 预览效果，确认无误后再执行。

删除消费者组

# 删除消费者组（仅当组内无活跃消费者时才能删除）
bin/kafka-consumer-groups.sh \
    --bootstrap-server kafka-01:9092 \
    --group my-consumer-group \
    --delete

ACL 权限管理

Kafka 的 ACL（Access Control List）根据客户端主体进行授权，主体可来自 SASL 认证或 SSL 客户端证书等机制。本节使用 SASL 示例，常用方式为 SASL/PLAIN 或 SASL/SCRAM。

开启 ACL 的 Broker 配置

# server.properties
# 使用 SASL_PLAINTEXT 协议（生产建议 SASL_SSL）
listeners=SASL_PLAINTEXT://kafka-01:9092
advertised.listeners=SASL_PLAINTEXT://kafka-01:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

# 使用 ZooKeeper 模式的 AclAuthorizer；KRaft 模式需采用对应的 Authorizer 配置
authorizer.class.name=kafka.security.authorizer.AclAuthorizer

# Super user（跳过所有 ACL 检查）
super.users=User:admin

# 当没有匹配的 ACL 规则时，默认拒绝（安全模式）
allow.everyone.if.no.acl.found=false

Producer 权限授权

# 允许用户 app-producer 向 orders Topic 发送消息
bin/kafka-acls.sh \
    --bootstrap-server kafka-01:9092 \
    --command-config admin-client.properties \
    --add \
    --allow-principal User:app-producer \
    --operation Write \
    --operation Describe \
    --topic orders

# Producer 通常需要 Write + Describe 权限
# Write：发送消息
# Describe：获取 Topic 元数据（必须，否则生产者无法发现分区信息）

Consumer 权限授权

# 允许用户 app-consumer 从 orders Topic 消费消息，并加入 my-group 消费者组
bin/kafka-acls.sh \
    --bootstrap-server kafka-01:9092 \
    --command-config admin-client.properties \
    --add \
    --allow-principal User:app-consumer \
    --operation Read \
    --operation Describe \
    --topic orders

# Consumer 还需要 Consumer Group 权限
bin/kafka-acls.sh \
    --bootstrap-server kafka-01:9092 \
    --command-config admin-client.properties \
    --add \
    --allow-principal User:app-consumer \
    --operation Read \
    --group my-consumer-group

查看与删除 ACL

# 查看所有 ACL
bin/kafka-acls.sh \
    --bootstrap-server kafka-01:9092 \
    --command-config admin-client.properties \
    --list

# 查看某 Topic 的 ACL
bin/kafka-acls.sh \
    --bootstrap-server kafka-01:9092 \
    --command-config admin-client.properties \
    --list \
    --topic orders

# 删除 ACL
bin/kafka-acls.sh \
    --bootstrap-server kafka-01:9092 \
    --command-config admin-client.properties \
    --remove \
    --allow-principal User:app-producer \
    --operation Write \
    --topic orders

配额管理

Kafka 配额（Quota）用于限制单个客户端（Producer 或 Consumer）的带宽使用，防止某个业务占用过多集群资源。

Producer / Consumer 带宽配额

# 限制用户 app-producer 的生产带宽为 10 MB/s
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type users \
    --entity-name app-producer \
    --alter \
    --add-config producer_byte_rate=10485760

# 限制用户 app-consumer 的消费带宽为 20 MB/s
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type users \
    --entity-name app-consumer \
    --alter \
    --add-config consumer_byte_rate=20971520

# 同时限制某个 Client ID 的生产和消费带宽
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type clients \
    --entity-name my-client \
    --alter \
    --add-config producer_byte_rate=5242880,consumer_byte_rate=10485760

# 查看配额配置
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type users \
    --entity-name app-producer \
    --describe

# 删除该用户的显式配额配置；是否仍限速取决于默认配额及其他适用配额
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type users \
    --entity-name app-producer \
    --alter \
    --delete-config producer_byte_rate

请求速率配额

# 限制某客户端的请求 CPU 使用率（以 Broker 单线程 100% 为基准的百分比）
bin/kafka-configs.sh \
    --bootstrap-server kafka-01:9092 \
    --entity-type users \
    --entity-name heavy-user \
    --alter \
    --add-config request_percentage=25

掌握检查与练习题

掌握检查

[ ] 能够独立完成 Kafka 集群的水平扩容，包括新 Broker 接入和分区重分配
[ ] 理解 kafka-reassign-partitions.sh 的 --generate、--execute、--verify 三个阶段
[ ] 掌握 Preferred Leader 选举的触发方式和自动平衡配置
[ ] 能够安全地执行滚动重启，并知道每步的等待条件
[ ] 会使用 kafka-consumer-groups.sh 查看 Lag，并能重置偏移量到指定时间点
[ ] 了解 Kafka ACL 的 Producer 和 Consumer 权限模板

练习题

扩容规划题：当前集群有 3 个 Broker（ID: 1,2,3），共 30 个 Topic，总分区数 300，副本因子均为 3。现在新增 3 台 Broker（ID: 4,5,6），请描述完整的分区重分配流程，以及如何通过限速、低峰执行和持续监控降低迁移对生产的影响，并说明不能保证完全无影响的条件。
偏移量重置题：消费者组 payment-processor 因为代码 Bug 错误处理了 2024-03-01 00:00:00 到 2024-03-01 06:00:00 之间的订单消息，需要回溯重处理这段时间的数据，请给出操作命令。
权限设计题：为新业务系统设计 Kafka 权限方案：该系统需要向 orders、payments 两个 Topic 写入消息，同时从 inventory Topic 读取消息（消费者组 new-biz-group），用户名为 new-biz-user，给出完整的 ACL 配置命令。

相关文档：性能调优实战 | 监控与故障排查 | 偏移量管理