卡夫卡的藏书阁【Book9】- Kafka Partition Reassign

“The meaning of life is that it stops.”
― Franz Kafka
生命的意义在於它会停止

如果今天有需求将某几个 topic 全部搬移到指定的某几个 broker，kafka 有提供脚本 kafka-reassign-partitions.sh 可以批次搬移

1. 在开始前先确认目前 topic 的状态

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: topicWithThreeBroker	Partition: 1	Leader: 1	Replicas: 0,1	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 2	Leader: 1	Replicas: 1,2	Isr: 1,2

2. 首先需要新增一个 json 档，填入想要重新分配 partition 的 topic

$ vim topics-to-move.json

新增内容如下：
{"topics": [{"topic": "topicWithThreeBroker"}],
"version":1
}

如果有多个topic用逗号分开即可

{"topics": [
    {"topic": "topicWithThreeBroker"},
    {"topic": "anotherTopic"}
],
"version":1
}

3. 产生 partition 分配表

参数说明：
topics-to-move-json-file => 填入我们刚刚产生的 json 档案名称
broker-list => 可以指定要分配到哪几个 broker
generate => 会显示当前 topic 的 partition 分配，并且自动产生我们申请要重新分配 partition 的 topic 的 partition 分配表

$ kafka-reassign-partitions --bootstrap-server localhost:9092 --topics-to-move-json-file topics-to-move.json --broker-list "0,1," --generate

Current partition replica assignment
{"version":1,"partitions":[{"topic":"topicWithThreeBroker","partition":0,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":1,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":2,"replicas":[0,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":5,"replicas":[0,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":6,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":7,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":8,"replicas":[0,2],"log_dirs":["any","any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"topicWithThreeBroker","partition":0,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":2,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":3,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":4,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":5,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":6,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":7,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":8,"replicas":[1,0],"log_dirs":["any","any"]}]}

参数说明：
topic => 指定的 topic
partition => 指定的 partition
replicas => 指定 partition 分配到哪些 broker
log_dirs => 指定 broker 的 log 路径，预设的 any 代表 broker 可以自由地去选择 replica 要放在哪里，目前 broker 实作选择 log 路径的方法是 round-robin alogorithn

将 current partition 的部分备份起来 partition_replica_assignment_backup.json，以防需要rollback

将 Proposed partition 的部分存到 expand_cluster_reassignment.json

4. 用指令verify查看此次将会有什麽变动

$ kafka-reassign-partitions --bootstrap-server localhost:9092 --reassignment-json-file expand_cluster_reassignment.json --verify

Status of partition reassignment:
Reassignment of partition topicWithThreeBroker-0 is complete.
There is no active reassignment of partition topicWithThreeBroker-1, but replica set is 2,1 rather than 0,1.
There is no active reassignment of partition topicWithThreeBroker-2, but replica set is 0,2 rather than 1,0.
There is no active reassignment of partition topicWithThreeBroker-3, but replica set is 1,2 rather than 0,1.
There is no active reassignment of partition topicWithThreeBroker-4, but replica set is 2,1 rather than 1,0.
There is no active reassignment of partition topicWithThreeBroker-5, but replica set is 0,2 rather than 0,1.
There is no active reassignment of partition topicWithThreeBroker-6, but replica set is 1,2 rather than 1,0.
There is no active reassignment of partition topicWithThreeBroker-7, but replica set is 2,1 rather than 0,1.
There is no active reassignment of partition topicWithThreeBroker-8, but replica set is 0,2 rather than 1,0.

Clearing broker-level throttles on brokers 0,1,2
Clearing topic-level throttles on topic topicWithThreeBroker

可以看到此次各 partition 的 replica 会怎麽变动

5. 执行重新分配 replica

$ kafka-reassign-partitions --bootstrap-server localhost:9092 --reassignment-json-file expand_cluster_reassignment.json --execute

Current partition replica assignment

{"version":1,"partitions":[{"topic":"topicWithThreeBroker","partition":0,"replicas":[1,0],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":1,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":2,"replicas":[0,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":5,"replicas":[0,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":6,"replicas":[1,2],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":7,"replicas":[2,1],"log_dirs":["any","any"]},{"topic":"topicWithThreeBroker","partition":8,"replicas":[0,2],"log_dirs":["any","any"]}]}

Save this to use as the --reassignment-json-file option during rollback
Successfully started partition reassignments for topicWithThreeBroker-0,topicWithThreeBroker-1,topicWithThreeBroker-2,topicWithThreeBroker-3,topicWithThreeBroker-4,topicWithThreeBroker-5,topicWithThreeBroker-6,topicWithThreeBroker-7,topicWithThreeBroker-8

6. 这时可以用verify检查执行状态，执行中会显示is in progress、执行完毕会显示 is complete

$ kafka-reassign-partitions --bootstrap-server localhost:9092 --reassignment-json-file expand_cluster_reassignment.json --verify

Status of partition reassignment:
Reassignment of partition topicWithThreeBroker-0 is complete.
Reassignment of partition topicWithThreeBroker-1 is complete.
Reassignment of partition topicWithThreeBroker-2 is complete.
Reassignment of partition topicWithThreeBroker-3 is complete.
Reassignment of partition topicWithThreeBroker-4 is complete.
Reassignment of partition topicWithThreeBroker-5 is complete.
Reassignment of partition topicWithThreeBroker-6 is complete.
Reassignment of partition topicWithThreeBroker-7 is complete.
Reassignment of partition topicWithThreeBroker-8 is complete.

Clearing broker-level throttles on brokers 0,1,2
Clearing topic-level throttles on topic topicWithThreeBroker

最後看一下 topic 是否有重新分配：

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 9	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 1	Leader: 0	Replicas: 0,1	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 2	Leader: 1	Replicas: 1,0	Isr: 0,1
	Topic: topicWithThreeBroker	Partition: 3	Leader: 1	Replicas: 0,1	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 4	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 5	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: topicWithThreeBroker	Partition: 6	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 7	Leader: 0	Replicas: 0,1	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 8	Leader: 0	Replicas: 1,0	Isr: 0,1

最後可以看到 partition 确实平均分配到 broker0, broker1之上，这个功能在系统扩增阶段是有机会用到的。

资料来源：
官网文件Link: Apache Kafka