Kafka在0.8以前版本中是没有High Available的，每个Partition没有Replication，一旦机器宕机或者某个Broker停止工作，受影响的Partition会变为不可用的，这对整个系统的可用性造成了影响。且集群越大，出现异常的几率就越大。

为什么需要Leader选举机制？

       引入HA后，每个Partition有多个副本Replica，需要再这些Replica中选举出一个Leader，Producer和Consumer只与这个Leader进行交互，其他的Replica与这个Leader进行数据的同步，从Leader中复制数据。
       想象一个情况，如果没有Leader，Partition的Replica间的数据一致性怎么保证？如果其中的一个Replica发生了宕机，怎么保证其他的Replica能继续提供服务而不造成数据丢失？在没有Leader的情况下，多个Replica间需要互相同步数据，这样就需要N*N条通路，一致性非常难以保证。
如果Replica中有一个Leader，其余的Replica只需要向此Leader进行数据的Fetch，因此只需要N条通路，简单且高效。

HA的设计

       Kafka应尽量将Replica分散到整个集群上。典型的做法是创建一个Topic时，使其Partition数量大于Broker数量，同时Replica尽量分散到不同机器（确保一台Broker宕机的影响最小）。
       分配Replica算法如下：
1.将n个broker和partition排序
2.第i个partition分配到第（i mod n）个broker上
3.第i个partition的第j个replica分配到第（(i + j) mod n）个broker上

消息的传播

       Producer通过Zookeeper找到某Paritition的Leader，将消息发送给Leader，Leader将其写入本地Log。其他的Replica（Follower）会从Leader处拉取消息，Flllower收到消息后（并不需要写入Log，也就是在内存中，提高了吞吐量），立即向Leader发送ACK，若Leader收到了所有Replica的ACK，该消息被认为是commit的。这是Consumer才能看到这条消息。

需要保证多少个备份？

       Leader中保存着与其同步的Replica的列表，称为ISR（in-sync Replica），如果一个Follower宕机或落后太多（可配置），Leader将其从ISR中移除。
       一条消息只有被ISR里的素有Follower都从Leader复制过去才认为已提交。

如何选举Leader？

       Kafka在Zookeeper中动态维护了一个ISR（in-sync replicas），其中所有的Replica都跟上了Leader，只有ISR中的成员才能被选为Leader。
       Kafka会在所有broker中选出一个controller，所有Partition的Leader选举都由controller决定。
       每个Partition会有多个备份，我们称某个Partition的Replica列表为AR（Assigned Replicas），AR中的第一个Replica称为“Preferred Replica”，它会被选为Leader，且Kafka保证Preferred Replica被均匀分布到所有Broker上，这样可以均衡集群的读写压力。

Kafka均衡Leader工具

$Kafka_HOME/bin/kafka-preferred-replica-election.sh --zookeeper localhost:2181
       broker宕机后，集群的leader可能处于一种不均衡的状态，此工具用来重新分配Leader所在Broker，使其均衡。

Kafka分区再均衡工具

       上一个工具只能在Partition的AR范围内调整其Leader，使Leader分布均匀，但这是不够的，此工具还可以调整Partition的AR。
       在对Kafka集群扩容的时候，已有的Topic不会自动将Partition迁移到新加入的Broker上，这时就需要用到此工具。

       此工具有三种用法：

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

       例子：
（1）generate
$KAFKA_HOME/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file /tmp/topics-to-move.json --broker-llist "4,5,6,7" --generate

（2）execute
$KAFKA_HOME/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file /tmp/reassign-plan.json --execute

（3）verify
$KAFKA_HOME/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file /tmp/reassign-plan.json --verify