第一步 配置新得broker

1. 將現有的叢集上任一個伺服器上的kafka目錄拷貝到新的伺服器上
2. 修改config/server.properties中的broker.id、log.dirs、listeners
3. 建立logs.dirs指定的目錄，並設定讀寫許可權(chomd -R 777 XXX)

broker.id=3
log.dirs=kafka-logs
listeners=PLAINTEXT://172.16.49.174:9092

第二步 啟動新的broker

bin/kafka-server-start.sh config/server.properties  & 
 

第三步 遷移指定topic的資料到新的broker

雖然經過上面兩個步驟後已經完成了叢集的擴容；但是叢集上原有的topic的資料不會自動遷移到新的broker上。可以在新的broker所在的伺服器上通過 ls /home/lxh/kafka_2.11-0.10.0.0/kafka-logs 檢視到並沒有一原有的topic名稱的檔案目錄（因為建立topic後會在config/server.properties中的配置的log.dirs 目錄中生產以topic名稱+分割槽編號的檔案目錄）；那麼就需要手動的區遷移資料

(一)、生成遷移分配規則json檔案

建立編輯要遷移的topic的 json檔案

vi topic-to-move.json  

比如要將topic名稱為test和paritioer_test的資料重新平衡到叢集中，就可以新增以下內容

{"topics": [{"topic": "test"},
            {"topic": "paritioer_test"}],
 "version":1
}

生成遷移分配規則json檔案

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topic-to-move.json --broker-list
 
 "0,1,2" --generate

得到的結果為

Current partition replica assignment 

{"version":1,"partitions":[{"topic":"test","partition":4,"replicas":[0,1]},{"topic":"test","partition":1,"replicas":[1,0]},{"topic":"paritioer_test","partition":0,"replicas":[0]},{"topic":"test","partition":2,"replicas":[0,1]},{"topic":"test","partition":0,"replicas":[0,1]},{"topic":"test","partition":3,"replicas":[1,0]}]}
Proposed partition reassignment configuration

{"version":1,"partitions":[{"topic":"test","partition":4,"replicas":[1,0]},{"topic":"test","partition":1,"replicas":[1,2]},{"topic":"test","partition":2,"replicas":[2,0]},{"topic":"paritioer_test","partition":0,"replicas":[0]},{"topic":"test","partition":0,"replicas":[0,1]},{"topic":"test","partition":3,"replicas":[0,2]}]}

其中的Current partition replica assignment指的是遷移前的partition replica；Proposed partition reassignment configuration 指的就是遷移分配規則json。需要將該json檔案儲存到json檔案中(如expand-cluster-reassignment.json)

(二)、執行遷移分配

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --execute

注意：在遷移過程中不能人為的結束或停止kafka服務，不然會有資料不一致的問題

(三)、驗證分配

在執行的過程中，可以新開一個終端執行以下命令來檢視執行是否正確完成

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --verify

輸出

Status of partition reassignment:
Reassignment of partition [test,4] completed successfully
Reassignment of partition [test,0] completed successfully
Reassignment of partition [paritioer_test,0] completed successfully
Reassignment of partition [test,3] completed successfully
Reassignment of partition [test,2] completed successfully
Reassignment of partition [test,1] completed successfully

在遷移完成過程後，可以使用以下命令看下topic的每個partitions的分佈情況

bin/kafka-topics.sh --zookeeper 172.16.49.173:2181 --describe --topic test

Topic:test  PartitionCount:5    ReplicationFactor:2 Configs:
    Topic: test Partition: 0    Leader: 0   Replicas: 0,1   Isr: 0,1
    Topic: test Partition: 1    Leader: 1   Replicas: 1,2   Isr: 1,2
    Topic: test Partition: 2    Leader: 2   Replicas: 2,0   Isr: 0,2
    Topic: test Partition: 3    Leader: 0   Replicas: 0,2   Isr: 0,2
    Topic: test Partition: 4    Leader: 0   Replicas: 1,0   Isr: 0,1

可以看到名為test的topic的有的資料以及存在於編號為2的新broker上了