Kafka Broker 运维
Kafka Broker 工作流程、ZooKeeper 元数据观察、节点服役退役和副本迁移操作。
Kafka Broker 运维
1、Kafka Broker工作流程
1.1 Zookeeper存储的Kafka信息
1）启动Zookeeper客户端
[root@kafka102 zookeeper-3.5.9]$ bin/zkCli.sh
or
[root@kafka102 zookeeper-3.5.9]$ bin/zkCli.sh -server kafka102:2181
2）通过ls命令可以查看kafka相关信息
[zk: kafka102:2181(CONNECTED) 1] ls /kafka
[admin, brokers, cluster, config, consumers, controller, controller_epoch, feature, isr_change_notification, latest_producer_id_block, log_dir_event_notification]
| 节点路径 | 核心作用 | 关键细节 | 补充说明 |
|---|---|---|---|
/kafka/admin | 集群管理操作的任务记录与状态存储 | 1. 子节点 delete_topics：记录待删除的 Topic 及删除任务状态，保证删除操作幂等性；2. 子节点 reassign_partitions：存储分区重分配的计划、进度，中断后可续跑；3. 子节点 preferred_replica_election：记录优先副本选举任务状态。 | 1./kafka/admin/delete_topics 仅在删除过程中临时存在，删除完成后节点会被自动清理；若删除速度快（数据量小、负载低），可能观察不到该节点；<br />2. reassign_partitions 和 preferred_replica_election 为 “按需创建” 节点，无对应操作时不会显示，属于正常现象。- |
/kafka/brokers | 存储集群所有 Broker 及 Topic 分区的核心元数据（最核心节点） | 1. 子节点 ids：列出所有在线 Broker 的 ID，及 Broker 的地址、端口、版本等信息；2. 子节点 topics：存储所有 Topic 的元数据（分区数、副本分配、ISR 列表等）；3. 子节点 seqid：自动生成 Broker ID 的自增序列（未手动指定 Broker ID 时生效）。 | |
/kafka/cluster | 存储集群级全局标识 | 核心为 id 子节点：记录集群唯一 ID，用于区分不同 Kafka 集群，避免元数据混淆。 | - |
/kafka/config | 存储动态配置（无需重启集群生效） | 1. 子节点 topics：各 Topic 的自定义配置（如 retention.ms、segment.bytes）；2. 子节点 brokers：各 Broker 的动态配置（如 listeners、log.retention.hours）；3. 子节点 clients/users：客户端配额、用户认证相关配置。 | - |
/kafka/consumers | 兼容旧版消费者（0.9 之前）的消费偏移量存储 | 新版消费者（0.9+）默认将偏移量存在 Kafka 内部 Topic __consumer_offsets，此节点仅用于兼容历史业务，通常为空。 | - |
/kafka/controller | 记录当前集群的 Controller Broker 信息 | 存储当前 Controller 的 Broker ID，是集群的 “指挥中心”，负责 Leader 选举、Broker 上下线管理；节点为临时节点，Controller 宕机后自动删除，触发新选举。 | - |
/kafka/controller_epoch | Controller 纪元号，防止 “脑裂” | 每次 Controller 重新选举，纪元号自增 1；确保只有最新纪元的 Controller 指令被集群执行，避免多 Controller 冲突。 | - |
/kafka/feature | 集群功能特性管理（兼容版本升级） | 存储各 Broker 支持的功能版本（如幂等性、事务），保证滚动升级时功能兼容，避免版本不一致导致异常。 | - |
/kafka/isr_change_notification | ISR 列表变更通知节点 | 当分区的同步副本（ISR）列表变化（如副本滞后被踢出），Controller 会在此写入通知，触发监控 / 告警或数据同步逻辑。 | - |
/kafka/latest_producer_id_block | 幂等 / 事务生产者的 PID 分配 | 存储最后一次分配的 Producer ID（PID）块，保证每个幂等 / 事务生产者的 PID 全局唯一，避免重复。 | - |
/kafka/log_dir_event_notification | 日志目录故障通知节点 | 当 Broker 磁盘日志目录异常（满、损坏），在此写入事件，Controller 感知后触发分区迁移等容灾操作。 | - |
模拟Kafka上下线，Zookeeper中数据变化
1）查看/kafka/brokers/ids路径上的节点
[zk: localhost:2181(CONNECTED) 2] ls /kafka/brokers/ids[0, 1, 2]
2）查看/kafka/controller路径上的数据
[zk: localhost:2181(CONNECTED) 5] get /kafka/controller
{"version":1,"brokerid":1,"timestamp":"1771677041573"}
cZxid = 0x8000001a3
ctime = Sat Feb 21 20:30:41 CST 2026
mZxid = 0x8000001a3
mtime = Sat Feb 21 20:30:41 CST 2026
pZxid = 0x8000001a3
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x29c7ff0ab5c0003
dataLength = 54
numChildren = 0
3）查看/kafka/brokers/topics/first/partitions/0/state路径上的数据
[zk: localhost:2181(CONNECTED) 6] get /kafka/brokers/topics/first/partitions/0/state
{"controller_epoch":10,"leader":0,"version":1,"leader_epoch":0,"isr":[0,2]}
cZxid = 0x80000021c
ctime = Sat Feb 21 22:39:05 CST 2026
mZxid = 0x80000021c
mtime = Sat Feb 21 22:39:05 CST 2026
pZxid = 0x80000021c
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 75
numChildren = 0
4）停止kafka104上的kafka
[root@kafka104 kafka-3.2.1]$ bin/kafka-server-stop.sh
5）再次查看/kafka/brokers/ids路径上的节点
[zk: localhost:2181(CONNECTED) 3] ls /kafka/brokers/ids[0, 1]
6）再次查看/kafka/controller路径上的数据
[zk: localhost:2181(CONNECTED) 15] get /kafka/controller{"version":1,"brokerid":0,"timestamp":"1637292471777"}
7）再次查看/kafka/brokers/topics/first/partitions/0/state路径上的数据
[zk: localhost:2181(CONNECTED) 16] get /kafka/brokers/topics/first/partitions/0/state {"controller_epoch":24,"leader":0,"version":1,"leader_epoch":18,"isr":[0,1]}
8）启动kafka104上的kafka
[root@kafka104 kafka-3.2.1]$ bin/kafka-server-start.sh -daemon ./config/server.properties
9）再次观察
再次观察1）、2）、3）步骤中的内容
2、生产经验--节点服役和退役
2.1 服役新节点
1）新节点准备（环境准备）
（1）关闭kafka104，并右键执行克隆操作
（2）开启kafka105，并修改IP地址
[root@kafka105 ~]# vim /etc/sysconfig/network-scripts/ifcfg-ens33
DEVICE=ens33
TYPE=Ethernet
ONBOOT=yes
BOOTPROTO=static
NAME="ens33"
IPADDR=192.168.2.105
PREFIX=24
GATEWAY=192.168.2.2
DNS1=192.168.10.2
（3）在kafka105上，修改主机名称为kafka105
（4）修改kafka105中kafka的broker.id为3
（5）删除kafka105中kafka下的datas和logs
[root@kafka105 kafka-3.2.1]$ rm -rf datas/* logs/*
（6）启动kafka102、kafka103、kafka104上的kafka集群
[root@kafka102 ~]$ zk.sh start
[root@kafka102 ~]$ kf.sh start
（7）单独启动kafka105中的kafka服务
在启动kafka105节点的kafka之前，请先查看已启动的kafka集群是否有topic，如果没有，请提前创建一个topic，2个备份，3个分区
1.创建主题
bin/kafka-topics.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --create --replication-factor 2 --partitions 3 --topic first
2.查看主题
[root@kafka102 kafka-3.2.1]$ xkafka-topic-list.sh
first
3.查看详情
[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh first
Topic: first TopicId: wNbSexjDQ8S96Y4ov1hSOg PartitionCount: 3 ReplicationFactor: 2 Configs: segment.bytes=1073741824
Topic: first Partition: 0 Leader: 2 Replicas: 2,1 Isr: 2,1
Topic: first Partition: 1 Leader: 1 Replicas: 1,0 Isr: 1,0
Topic: first Partition: 2 Leader: 0 Replicas: 0,2 Isr: 0,2
4.启动105
[root@kafka105 kafka-3.2.1]$ bin/kafka-server-start.sh -daemon config/server.properties
2）对已存在的topic执行负载均衡操作
（1）创建一个要均衡的主题
[root@kafka102 kafka-3.2.1]$ touch topics-to-move.json
[root@kafka102 kafka-3.2.1]$ vim topics-to-move.json
{
"topics": [
{"topic": "first"}
],
"version": 1
}
（2）生成一个负载均衡的计划
[root@kafka102 kafka-3.2.1]# bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2,3" --generate
Current partition replica assignment
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[2,0],"log_dirs":["any","any"]}]}
Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,3],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[3,1],"log_dirs":["any","any"]}]}
（3）创建副本存储计划（所有副本存储在broker0、broker1、broker2、broker3中）
[root@kafka102 kafka-3.2.1]$ vim increase-replication-factor.json
（4）执行副本存储计划
[root@kafka102 kafka-3.2.1]$ bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092,kafka103:9092,kafka104:9092 --reassignment-json-file increase-replication-factor.json --execute
（5）验证副本存储计划
[root@kafka102 kafka-3.2.1]$ bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092 --reassignment-json-file increase-replication-factor.json --verify
（6）查看详情
[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh first
Topic: first TopicId: wNbSexjDQ8S96Y4ov1hSOg PartitionCount: 3 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: first Partition: 0 Leader: 2 Replicas: 2,3,0 Isr: 2,0,3
Topic: first Partition: 1 Leader: 3 Replicas: 3,0,1 Isr: 1,0,3
Topic: first Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 0,2,1
(7):作业
完成节点服役2.2退役旧节点
主要思想逻辑
1）执行负载均衡操作
2）关闭kafka105节点上的kafka服务
先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。
1）创建一个要均衡的主题
[root@kafka102 kafka-3.2.1]$ vim topics-to-move.json
{
"topics": [
{"topic": "first"}
],
"version": 1
}
2）查看topic详情
[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh first
Topic: first TopicId: wNbSexjDQ8S96Y4ov1hSOg PartitionCount: 3 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: first Partition: 0 Leader: 2 Replicas: 2,3,0 Isr: 2,0,3
Topic: first Partition: 1 Leader: 3 Replicas: 3,0,1 Isr: 1,0,3
Topic: first Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 0,2,1
3）创建执行计划
bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate
----
Current partition replica assignment
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[1,3],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[3,1],"log_dirs":["any","any"]}]}
Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"first","partition":1,"replicas":[0,2],"log_dirs":["any","any"]},{"topic":"first","partition":2,"replicas":[1,0],"log_dirs":["any","any"]}]}
4）创建副本存储计划（所有副本存储在broker0、broker1、broker2中）
[root@kafka102 kafka-3.2.1]$ vim increase-replication-factor.json
{
"version": 1,
"partitions": [{
"topic": "first",
"partition": 0,
"replicas": [2, 0, 1],
"log_dirs": ["any", "any", "any"]
}, {
"topic": "first",
"partition": 1,
"replicas": [0, 1, 2],
"log_dirs": ["any", "any", "any"]
}, {
"topic": "first",
"partition": 2,
"replicas": [1, 2, 0],
"log_dirs": ["any", "any", "any"]
}]
}
5）执行副本存储计划
[root@kafka102 kafka-3.2.1]$ bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092 --reassignment-json-file increase-replication-factor.json --execute
Current partition replica assignment
{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2,3,0],"log_dirs":["any","any","any"]},{"topic":"first","partition":1,"replicas":[3,0,1],"log_dirs":["any","any","any"]},{"topic":"first","partition":2,"replicas":[0,1,2],"log_dirs":["any","any","any"]}]}
Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for first-0,first-1,first-2
6）验证副本存储计划
[root@kafka102 kafka-3.2.1]$ bin/kafka-reassign-partitions.sh --bootstrap-server kafka102:9092 --reassignment-json-file increase-replication-factor.json --verify
Status of partition reassignment:
Reassignment of partition first-0 is complete.
Reassignment of partition first-1 is complete.
Reassignment of partition first-2 is complete.
Clearing broker-level throttles on brokers 0,1,2,3
Clearing topic-level throttles on topic first
7）查看topic详情
[root@kafka102 kafka-3.2.1]$ xkafka-topic-describe.sh first
Topic: first TopicId: wNbSexjDQ8S96Y4ov1hSOg PartitionCount: 3 ReplicationFactor: 3 Configs: segment.bytes=1073741824
Topic: first Partition: 0 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: first Partition: 1 Leader: 0 Replicas: 0,1,2 Isr: 1,0,2
Topic: first Partition: 2 Leader: 0 Replicas: 1,2,0 Isr: 0,2,1
8）关闭kafka105节点上的kafka服务
[root@kafka105 kafka-3.2.1]$ bin/kafka-server-stop.sh