kafka叢集
broker 0（10.11.32.76    hadooptest76.bj）
broker 1（10.11.32.77    hadooptest77.bj）
broker 2（10.11.32.81    hadooptest81.bj）

操作任務：下線broker 2節點，確保業務不中斷，資料不丟失。

一、先決條件，kafka啟動必須加上JMX_PORT=9999(自己選一個空閒埠)
在zookeeper的客戶端使用get  /kafka/brokers/ids/[0,1,2](一個數字代表你的brokerid),如果返回資訊中的jmx_port 值為-1就說明啟動時沒有指定JMX_PORT，這樣不需要進行下面步驟，請直接重啟叢集加上配置,並且server.properties需要設定auto.create.topics.enable=false。
auto.create.topics.enable 預設值：true
是否允許自動建立topic。如果設為true，那麼produce，consume或者fetch metadata一個不存在的topic時，就會自動建立一個預設replication factor和partition number的topic。

配置jmx服務
方法一：（網上有文章讓這樣配置，但會報錯，請不要這樣操作）
kafka server中預設是不啟動jmx埠的，需要使用者自己配置
 vim bin/kafka-run-class.sh
#最前面新增一行
JMX_PORT=8060
（
不要這樣新增不然會報錯：
Error: Exception thrown by the agent : java.rmi.server.ExportException: Port already in use: 8060; nested exception is:
        java.net.BindException: Address already in use (Bind failed)
）

方法二：（建議）
建議啟動kafka服務的時候前邊加上JMX_PORT：
JMX_PORT=9999 /usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-server-start.sh


我自定義一個啟動指令碼
cat start-kafka-all.sh
#!/bin/bash

while read i1 i2
do
ssh -T $i1 << TT
hostname
JMX_PORT=9999 /usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-server-start.sh /usr/local/kafka_2.9.2-0.8.1.1/config/server.properties >/usr/local/kafka_2.9.2-0.8.1.1/kafka.out &
TT
done < iplist

iplist檔案內容
cat iplist
10.11.32.76    hadooptest76.bj
10.11.32.77    hadooptest77.bj
10.11.32.81    hadooptest81.bj

二、移動topic
新建topic
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-topics.sh --create --topic test01 --partitions 6 --replication-factor 2 --zookeeper 10.11.32.76:2181

檢視：
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-topics.sh --list --zookeeper 10.11.32.76:2181
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-topics.sh --describe --zookeeper 10.11.32.76:2181 --topic test01

生產者：
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-console-producer.sh --broker-list 10.11.32.76:9092 --topic test01

消費者：
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-console-consumer.sh --zookeeper 10.11.32.76:2181 --topic test01 --from-beginning


建立topics-to-move.json檔案
[

[email protected] mv_topic]# cat topics-to-move.json
{"topics": [
 {"topic": "test01"},
 {"topic": "test02"},
 {"topic": "test03"}
 ],
 "version":1
}

#test01、test02、test03 為topic名字

生成移動指令碼
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-reassign-partitions.sh --zookeeper 10.11.32.76:2181 --topics-to-move-json-file topics-to-move.json --broker-list "0,1" --generate

這樣就會生成一串新的json資料Current partition replica assignment和Proposed partition reassignment configuration
Proposed partition reassignment configuration下面生成的就是將分割槽重新分佈到broker 0、1上的結果。
我們將這些內容儲存到名為result.json檔案裡面（檔名不重要，檔案格式也不一定要以json為結尾，只要保證內容是json即可），然後執行這些reassign plan：
{"version":1,"partitions":[{"topic":"fortest1","partition":0,"replicas":[3,4]},其他部分省略}
將這一串json寫入新檔案reassignment-node.json中

例如：
[

[email protected] mv_topic]# /usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-reassign-partitions.sh --zookeeper 10.111.32.76:2181 --topics-to-move-json-file topics-to-move.json --broker-list "0,1" --generate
Current partition replica assignment

{"version":1,"partitions":[{"topic":"test01","partition":2,"replicas":[2,0]},{"topic":"test01","partition":3,"replicas":[0,2]},{"topic":"test01","partition":5,"replicas":[2,1]},{"topic":"test02","partition":5,"replicas":[1,0]},{"topic":"test03","partition":1,"replicas":[2,0]},{"topic":"test01","partition":4,"replicas":[1,0]},{"topic":"test01","partition":0,"replicas":[0,1]},{"topic":"test02","partition":0,"replicas":[0,1]},{"topic":"test02","partition":4,"replicas":[0,1]},{"topic":"test02","partition":3,"replicas":[1,0]},{"topic":"test03","partition":2,"replicas":[0,1]},{"topic":"test03","partition":5,"replicas":[0,2]},{"topic":"test02","partition":2,"replicas":[0,1]},{"topic":"test01","partition":1,"replicas":[1,2]},{"topic":"test03","partition":4,"replicas":[2,1]},{"topic":"test03","partition":0,"replicas":[1,2]},{"topic":"test02","partition":1,"replicas":[1,0]},{"topic":"test03","partition":3,"replicas":[1,0]}]}
Proposed partition reassignment configuration

{"version":1,"partitions":[{"topic":"test01","partition":2,"replicas":[0,1]},{"topic":"test01","partition":3,"replicas":[1,0]},{"topic":"test01","partition":5,"replicas":[1,0]},{"topic":"test02","partition":5,"replicas":[0,1]},{"topic":"test03","partition":1,"replicas":[1,0]},{"topic":"test01","partition":4,"replicas":[0,1]},{"topic":"test01","partition":0,"replicas":[0,1]},{"topic":"test02","partition":0,"replicas":[1,0]},{"topic":"test02","partition":4,"replicas":[1,0]},{"topic":"test02","partition":3,"replicas":[0,1]},{"topic":"test03","partition":5,"replicas":[1,0]},{"topic":"test03","partition":2,"replicas":[0,1]},{"topic":"test02","partition":2,"replicas":[1,0]},{"topic":"test03","partition":4,"replicas":[0,1]},{"topic":"test01","partition":1,"replicas":[1,0]},{"topic":"test03","partition":0,"replicas":[0,1]},{"topic":"test02","partition":1,"replicas":[0,1]},{"topic":"test03","partition":3,"replicas":[1,0]}]}

新建vi reassignment-node.json檔案
[[email protected] mv_topic]# vi reassignment-node.json
{"version":1,"partitions":[{"topic":"test01","partition":2,"replicas":[0,1]},{"topic":"test01","partition":3,"replicas":[1,0]},{"topic":"test01","partition":5,"replicas":[1,0]},{"topic":"test02","partition":5,"replicas":[0,1]},{"topic":"test03","partition":1,"replicas":[1,0]},{"topic":"test01","partition":4,"replicas":[0,1]},{"topic":"test01","partition":0,"replicas":[0,1]},{"topic":"test02","partition":0,"replicas":[1,0]},{"topic":"test02","partition":4,"replicas":[1,0]},{"topic":"test02","partition":3,"replicas":[0,1]},{"topic":"test03","partition":5,"replicas":[1,0]},{"topic":"test03","partition":2,"replicas":[0,1]},{"topic":"test02","partition":2,"replicas":[1,0]},{"topic":"test03","partition":4,"replicas":[0,1]},{"topic":"test01","partition":1,"replicas":[1,0]},{"topic":"test03","partition":0,"replicas":[0,1]},{"topic":"test02","partition":1,"replicas":[0,1]},{"topic":"test03","partition":3,"replicas":[1,0]}]}


這時候,萬事俱備, 開始遷移：
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-reassign-partitions.sh --zookeeper 192.168.103.47:2181 --reassignment-json-file reassignment-node.json --execute
如果topic比較多的話需要多等待一些時間，等待完成調整，檢視下所有topic已經遷移到制定的broker上

過一會檢視topic
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test01

三、讓節點機器下線（必須要進行，及已經平衡了topic）
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 10.11.32.76:2181,10.11.32.77:2181,10.11.32.81:2181/kafka --broker 2 --num.retries 3 --retry.interval.ms 600

四、重新leader、重新均衡
下線節點完成後重新均衡下
/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-preferred-replica-election.sh --zookeeper 10.11.32.76:2181
成功後重新觀察tpoic狀態會發現Leader、Replicas都正常了

參考：
Kafka叢集擴充套件以及重新分佈分割槽
https://www.iteblog.com/archives/1611.html
官網
https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools#Replicationtools-1.ControlledShutdown
kafka的遷移乾貨
http://www.cnblogs.com/dycg/p/3922352.html
apache Kafka下線broker的操作
http://blog.csdn.net/lizhitao/article/details/42266327
Kafka叢集平滑重啟
http://blog.csdn.net/clarencezi/article/details/42271037
Kafka學習之四 Kafka常用命令
http://blog.csdn.net/code52/article/details/50935849
kafka server部署配置優化
http://blog.csdn.net/lizhitao/article/details/42180265
kafka擴容
http://zhouxinyu1991.blog.51cto.com/6095086/1876616