“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