kafka 迁移 broker

采用重新分配分区，将 broker.id 1/2/3 平滑迁移至 4/5/6

1. 新建节点（例如 4,5,6），并加入集群，确认 broker 都存活

   # 返回 [1, 2, 3, 4, 5, 6]
   /data/bkee/service/zk/bin/zkCli.sh -server zk.service.consul:2181 ls /common_kafka/brokers/ids

2. 找到所有 topic （也可以找单个 topic 试验）

   ./kafka-topics.sh --zookeeper zk.service.consul:2181/common_kafka --describe | egrep '^Topic:' | awk '{print $1}' | awk -F ':' '{print $2}' > topic.list

3. 将 topic 列表转换为 topics-to-move.json

   echo -n '{"topics": [' > topics-to-move.json
   for i in $(cat topic.list); do echo -n "{\"topic\": \"$i\"},"; done >> topics-to-move.json
   echo -n '], "version":1}' >> topics-to-move.json
   # 检验一下 json 格式
   cat topics-to-move.json | jq

4. 生成分区重分配 json，写入 generate.json

   # 确认数据不为空
   ./kafka-reassign-partitions.sh --zookeeper zk.service.consul:2181/common_kafka --topics-to-move-json-file 1.json --broker-list 4,5,6 --generate
   # 将 Proposed config 写入 generate.json
   ./kafka-reassign-partitions.sh --zookeeper zk.service.consul:2181/common_kafka --topics-to-move-json-file 1.json --broker-list 4,5,6 --generate | tail -n 1 > generate.json

5. 执行分区重分配

   ./kafka-reassign-partitions.sh --zookeeper zk.service.consul:2181/common_kafka --reassignment-json-file generate.json --execute

6. 验证执行状态

   # 输出整体状态
   ./kafka-reassign-partitions.sh --zookeeper zk.service.consul:2181/common_kafka --reassignment-json-file new.json --verify
   
   ./kafka-reassign-partitions.sh --zookeeper zk.service.consul:2181/common_kafka --reassignment-json-file new.json --verify | grep successfully
   
   # 输出为空即执行成功
   ./kafka-reassign-partitions.sh --zookeeper zk.service.consul:2181/common_kafka --reassignment-json-file new.json --verify | grep -v successfully

7. 关闭 broker.id 1/2/3 的进程，迁移完成