Kafka集群扩展以及重新分布分区

我们往已经部署好的Kafka集群里面添加机器是最正常不过的需求，而且添加起来非常地方便，我们需要做的事是从已经部署好的Kafka节点中复制相应的配置文件，然后把里面的broker id修改成全局唯一的，最后启动这个节点即可将它加入到现有Kafka集群中。

　　但是问题来了，新添加的Kafka节点并不会自动地分配数据，所以无法分担集群的负载，除非我们新建一个topic。但是现在我们想手动将部分分区移到新添加的Kafka节点上，Kafka内部提供了相关的工具来重新分布某个topic的分区。在重新分布topic分区之前，我们先来看看现在topic的各个分区的分布位置：

. /bin/kafka-topics .sh --topic iteblog --describe --zookeeper www.iteblog.com:2181

Topic:iteblog PartitionCount:7  ReplicationFactor:2 Configs:

   Topic: iteblog  Partition: 0  Leader: 1 Replicas: 1,2 Isr: 1,2

   Topic: iteblog  Partition: 1  Leader: 2 Replicas: 2,3 Isr: 2,3

   Topic: iteblog  Partition: 2  Leader: 3 Replicas: 3,4 Isr: 3,4

   Topic: iteblog  Partition: 3  Leader: 4 Replicas: 4,1 Isr: 4,1

   Topic: iteblog  Partition: 4  Leader: 1 Replicas: 1,3 Isr: 1,3

   Topic: iteblog  Partition: 5  Leader: 2 Replicas: 2,4 Isr: 2,4

   Topic: iteblog  Partition: 6  Leader: 3 Replicas: 3,1 Isr: 3,1

从上面的输出可以看出，iteblog主题一共有7个分区，但是我们broker的个数只有4个，所以会导致某些broker维护更多的分区。现在我们在现有集群的基础上再添加一个Kafka节点，然后使用Kafka自带的kafka-reassign-partitions.sh工具来重新分布分区。该工具有三种使用模式：

　　1、generate模式，给定需要重新分配的Topic，自动生成reassign plan（并不执行）
　　2、execute模式，根据指定的reassign plan重新分配Partition
　　3、verify模式，验证重新分配Partition是否成功

现在我们需要将原先分布在broker 1-4节点上的分区重新分布到broker 1-5节点上，借助kafka-reassign-partitions.sh工具生成reassign plan，不过我们先得按照要求定义一个文件，里面说明哪些topic需要重新分区，文件内容如下：

[[e-mail] ~]$ cat topics-to-move.json

{ "topics" : [{ "topic" : "iteblog" }],

  "version" :1

}

然后使用kafka-reassign-partitions.sh工具生成reassign plan

[[e-mail] ~]$ bin/kafka-reassign-partitions.sh --zookeeper www.iteblog.com:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2,3,4,5" --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"iteblog","partition":3,"replicas":[4,1]},{"topic":"iteblog","partition":5,"replicas":[2,4]},{"topic":"iteblog","partition":4,"replicas":[1,3]},{"topic":"iteblog","partition":0,"replicas":[1,2]},{"topic":"iteblog","partition":6,"replicas":[3,1]},{"topic":"iteblog","partition":1,"replicas":[2,3]},{"topic":"iteblog","partition":2,"replicas":[3,4]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"iteblog","partition":3,"replicas":[3,5]},{"topic":"iteblog","partition":5,"replicas":[5,3]},{"topic":"iteblog","partition":4,"replicas":[4,1]},{"topic":"iteblog","partition":0,"replicas":[5,2]},{"topic":"iteblog","partition":6,"replicas":[1,4]},{"topic":"iteblog","partition":1,"replicas":[1,3]},{"topic":"iteblog","partition":2,"replicas":[2,4]}]} 

Proposed partition reassignment configuration下面生成的就是将分区重新分布到broker 1-5上的结果。我们将这些内容保存到名为result.json文件里面（文件名不重要，文件格式也不一定要以json为结尾，只要保证内容是json即可），然后执行这些reassign plan：

[[e-mail] ~]$ bin/kafka-reassign-partitions.sh --zookeeper www.iteblog.com:2181 --reassignment-json-file result.json --execute Current partition replica assignment {"version":1,"partitions":[{"topic":"iteblog","partition":3,"replicas":[4,1]},{"topic":"iteblog","partition":5,"replicas":[2,4]},{"topic":"iteblog","partition":4,"replicas":[1,3]},{"topic":"iteblog","partition":0,"replicas":[1,2]},{"topic":"iteblog","partition":6,"replicas":[3,1]},{"topic":"iteblog","partition":1,"replicas":[2,3]},{"topic":"iteblog","partition":2,"replicas":[3,4]}]} Save this to use as the --reassignment-json-file option during rollback Successfully started reassignment of partitions {"version":1,"partitions":[{"topic":"iteblog","partition":1,"replicas":[1,3]},{"topic":"iteblog","partition":5,"replicas":[5,3]},{"topic":"iteblog","partition":4,"replicas":[4,1]},{"topic":"iteblog","partition":6,"replicas":[1,4]},{"topic":"iteblog","partition":2,"replicas":[2,4]},{"topic":"iteblog","partition":0,"replicas":[5,2]},{"topic":"iteblog","partition":3,"replicas":[3,5]}]} 

这样Kafka就在执行reassign plan，我们可以校验reassign plan是否执行完成：

[[e-mail] ~]$ bin/kafka-reassign-partitions.sh --zookeeper www.iteblog.com:2181 --reassignment-json-file result.json --verify Status of partition reassignment: Reassignment of partition [iteblog,1] completed successfully Reassignment of partition [iteblog,5] is still in progress Reassignment of partition [iteblog,4] completed successfully Reassignment of partition [iteblog,6] completed successfully Reassignment of partition [iteblog,2] completed successfully Reassignment of partition [iteblog,0] is still in progress Reassignment of partition [iteblog,3] completed successfully [[e-mail] ~]$ bin/kafka-reassign-partitions.sh --zookeeper www.iteblog.com:2181 --reassignment-json-file result.json --verify Status of partition reassignment: Reassignment of partition [iteblog,1] completed successfully Reassignment of partition [iteblog,5] completed successfully Reassignment of partition [iteblog,4] completed successfully Reassignment of partition [iteblog,6] completed successfully Reassignment of partition [iteblog,2] completed successfully Reassignment of partition [iteblog,0] completed successfully Reassignment of partition [iteblog,3] completed successfully 

可以看出，分区正在Reassignment的状态是still in progress；如果分区Reassignment完成则completed successfully，然后我们就可以看到分区已经按照生成的reassign plan进行，我们可以看下topic各个分区现在的分布情况：

[[e-mail] ~]$ ./bin/kafka-topics.sh --topic iteblog --describe --zookeeper www.iteblog.com:2181 Topic:iteblog PartitionCount:7 ReplicationFactor:2 Configs: Topic: iteblog Partition: 0 Leader: 5 Replicas: 5,2 Isr: 2,5 Topic: iteblog Partition: 1 Leader: 1 Replicas: 1,3 Isr: 3,1 Topic: iteblog Partition: 2 Leader: 2 Replicas: 2,4 Isr: 4,2 Topic: iteblog Partition: 3 Leader: 3 Replicas: 3,5 Isr: 3,5 Topic: iteblog Partition: 4 Leader: 1 Replicas: 4,1 Isr: 1,4 Topic: iteblog Partition: 5 Leader: 5 Replicas: 5,3 Isr: 3,5 Topic: iteblog Partition: 6 Leader: 1 Replicas: 1,4 Isr: 1,4 

分区的分布的确和操作之前不一样了，broker 5上已经有分区分布上去了。但是仔细的同学应该可以发现，broker 4上居然没有分区的Leader，这肯定不是我们想要的！所以使用kafka-reassign-partitions.sh工具生成的reassign plan只是一个建议，方便大家而已。其实我们自己完全可以编辑一个reassign plan，然后执行它，如下：

{

     "version" : 1,

     "partitions" : [

         {

             "topic" : "iteblog" ,

             "partition" : 0,

             "replicas" : [

                 1,

                 2

             ]

         },

         {

             "topic" : "iteblog" ,

             "partition" : 1,

             "replicas" : [

                 2,

                 3

             ]

         },

         {

             "topic" : "iteblog" ,

             "partition" : 2,

             "replicas" : [

                 3,

                 4

             ]

         },

         {

             "topic" : "iteblog" ,

             "partition" : 3,

             "replicas" : [

                 4,

                 5

             ]

         },

         {

             "topic" : "iteblog" ,

             "partition" : 4,

             "replicas" : [

                 5,

                 1

             ]

         },

         {

             "topic" : "iteblog" ,

             "partition" : 5,

             "replicas" : [

                 1,

                 3

             ]

         },

         {

             "topic" : "iteblog" ,

             "partition" : 6,

             "replicas" : [

                 2,

                 4

             ]

         }

     ]

}

将上面的json数据文件保存到result.json文件中，然后也是执行它：

[[e-mail] ~]$ bin/kafka-reassign-partitions.sh --zookeeper www.iteblog.com:2181 --reassignment-json-file result.json --execute 

等这个reassign plan执行完，我们再来看看分区的分布：

[[e-mail] ~]$ ./bin/kafka-topics.sh --topic iteblog --describe --zookeeper www.iteblog.com:2181 Topic:iteblog PartitionCount:7 ReplicationFactor:2 Configs: Topic: iteblog Partition: 0 Leader: 1 Replicas: 1,2 Isr: 2,1 Topic: iteblog Partition: 1 Leader: 2 Replicas: 2,3 Isr: 3,2 Topic: iteblog Partition: 2 Leader: 3 Replicas: 3,4 Isr: 4,3 Topic: iteblog Partition: 3 Leader: 4 Replicas: 4,5 Isr: 5,4 Topic: iteblog Partition: 4 Leader: 5 Replicas: 5,1 Isr: 1,5 Topic: iteblog Partition: 5 Leader: 1 Replicas: 1,3 Isr: 3,1 Topic: iteblog Partition: 6 Leader: 2 Replicas: 2,4 Isr: 4,2 

果然已经按照我们需求分布了。。