Broker_退役节点

将node1 broker节点退役，注意不能丢失数据。

1. 编写一个json文件，配置想要被负载均衡的主题（最好将被退役节点上有数据的主题都写上）

   xxxxxxxxxx
   [root@node2 kafka]# vim jsons/topics_to_move.json
   {
    "topics": [
    {"topic": "topicA"}
    ],
    "version": 1
   }
   

2. 根据1编写的json文件生成一个负载均衡的计划--broker-list 去掉要退役的节点

   xxxxxxxxxx
   [root@node2 kafka]# kafka-reassign-partitions.sh --bootstrap-server node2:9092 --topics-to-move-json-file jsons/topics_to_move.json --broker-list "0,1,2" --generate
   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"topicA","partition":0,"replicas":[3,0],"log_dirs":["any","any"]},{"topic":"topicA","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicA","partition":2,"replicas":[1,2],"log_dirs":["any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"topicA","partition":0,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"topicA","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicA","partition":2,"replicas":[1,2],"log_dirs":["any","any"]}]}
   

3. 创建副本存储计划（所有副本存储在 broker0、broker1、broker2中）

   xxxxxxxxxx
   [root@node2 kafka]# vim jsons/increase-replication-factor2.json
   {"version":1,"partitions":[{"topic":"topicA","partition":0,"replicas":[2,0],"log_dirs":["any","any"]},{"topic":"topicA","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicA","partition":2,"replicas":[1,2],"log_dirs":["any","any"]}]}
   

4. 执行副本存储计划

   xxxxxxxxxx
   [root@node2 kafka]# kafka-reassign-partitions.sh --bootstrap-server node2:9092 --reassignment-json-file jsons/increase-replication-factor2.json --execute
   

   提示信息：

   xxxxxxxxxx
   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"topicA","partition":0,"replicas":[3,0],"log_dirs":["any","any"]},{"topic":"topicA","partition":1,"replicas":[0,1],"log_dirs":["any","any"]},{"topic":"topicA","partition":2,"replicas":[1,2],"log_dirs":["any","any"]}]}
   Save this to use as the --reassignment-json-file option during rollback
   Successfully started partition reassignments for topicA-0,topicA-1,topicA-2
   

5. 验证副本存储计划

   xxxxxxxxxx
   [root@node2 kafka]# kafka-reassign-partitions.sh --bootstrap-server node2:9092 --reassignment-json-file jsons/increase-replication-factor2.json --verify
   

   结果信息：

   xxxxxxxxxx
   Status of partition reassignment:
   Reassignment of partition topicA-0 is complete.
   Reassignment of partition topicA-1 is complete.
   Reassignment of partition topicA-2 is complete.
   
   Clearing broker-level throttles on brokers 0,1,2,3
   Clearing topic-level throttles on topic topicA
   

6. zk节点中再次确认（步骤2中看哪个分区在broker-id=3上有数据）

   xxxxxxxxxx
   [zk: node4:2181(CONNECTED) 25] get /kafka/brokers/topics/topicA/partitions/0/state
   {"controller_epoch":2,"leader":2,"version":1,"leader_epoch":9,"isr":[0,2]}
   

7. 在node1上执行kafka停止命令

   xxxxxxxxxx
   [root@node1 kafka]# bin/kafka-server-stop.sh
   [root@node1 kafka]# jps
   4393 Jps