上篇文章说到，当leader宕机，这时候follower可以参与竞争leader，但不是所有follower都有资格，只有在ISR里面的follower副本才有资格，在0.9.0.0之前有两个参数，后面版本就一个超时时间，默认10s。Highwatermark 和 log end offset，在hw和leo之间的是未提交的消息在，这些消息是不会被消费，在hw之前的消息是已经同步到副本的消息，这些会被消费。
一个kafka集群中，某个broker被选举出来，即用controller来管理协调kafka集群。
每个kafka集群任意时刻都只能有一个controller。当集群启动时候，所有broker都会参与controller竞选，一旦controller崩溃，其他的broker会重新竞选。
Controller管理状态
一类是每台broker上的分区副本，一类是每个分区的leader副本信息。分为副本状态和分区状态。Controller为了管理这两个分别引入了管理副本状态 和 分区状态。
1）副本状态机（replica state machine）
Kafka为副本定义了7个状态以及每个状态流转规则。
NewReplica：controller创建副本时最初状态。当在这个状态时候，副本只能成为follower副本。
OnlineReplica：启动副本后变为该状态。在该状态下，副本既可以成为follower副本也可以成为leader副本。
OfflineReplica：一旦副本的broler崩溃，则副本变为该状态。
ReplicatDeletionStarted：若开启了topic删除操作，topic所有分区的所有副本都被删除。则进入该状态。
ReplicaDeletionSuccessful：副本成功响应了删除副本请求，则进入该状态。
ReplicaDeletionIneligible：若副本删除失败，则进入该状态。
NonExistentReplica：若副本成功删除，则进入改状态。
当创建某个topic后，该topic分区下所有副本都是NonExistent状态的，此时controller加载zookeeper中该topic每个分区的所有副本信息存到内存中，同时把状态改为new，之后controller选择该分区副本列表中第一个副本作为分区leader副本并设置所有副本进入ISR，然后在zookeeper中持久化该决定。
一旦确定ISR和leader后，controller会将所有副本状态设置为online，正常工作。
当开启了topic删除操作时，controller会尝试停止所有副本，此时副本停止向leader获取数据，若停止的副本是leader，则controller会设置分区leader为no_leader，之后副本为offline状态。一旦都进入到offline时，controller会将副本进一步变为replicaDeleteTionStarted表示删除开始。删除成功则successful，删除失败则Ineligible。
2）分区状态机（partition state machine）
除了副本状态，controller还引入了分区状态机来负责集群下所有分区管理状态。
NonExistent：表示不存在的分区或已删除的分区。
NewPartition：一旦被创建，分区便处于该状态。此时，kafka已经为分区确定了副本列表，但尚未选举出leader和ISR.
onlinePatition：一旦该分区的leader被选出，则进入该状态。这也是分区正常工作时候的状态。
OfflinePartition：在成功选举出leader后，若leader所在broker宕机，则分区将进入该状态，表明无法正常工作。
当创建topic时，controller负责创建分区对象，他首先会短暂将所有分区状态设置为nonExitent，之后马上读取zookeeper中的副本分配方案。然后令分区状态进入newPartition。处于newPartition状态分区尚未有leader和ISR，因此controller会初始化leader和ISR信息并设置分区状态为onlinePartition。
若用户发起删除topic操作或者关闭broker操作，那么controller会令受影响的分区进入offline状态，如果是删除操作，controller会开启分区下面副本操作，并设置状态为nonExistent，如果是关闭broker操作，则controller会判断是否是分区leader。
Controler职责
1、更新集群元数据信息。
2、创建topic。
3、删除topic。
4、分区重分配。
5、Preferred leader 副本选举。
6、Topic 分区扩展。
7、Broker加入集群。
8、Broker崩溃。
9、受控关闭。
10、Controller leader选举。
1）更新集群元数据信息
当分区信息发生变更，controller将变更后的信息封装进updateMetadataRequests请求，然后发送给集群中的每个broker，这样clients在请求数据时总会获取最新、最及时的分区信息。
2）创建topic
Controller启动时候会创建一个zookeeper的监听器，该监听器的任务就是监听topic节点变化。
A. 通过kafka-topics脚本--create创建。
B. 构造createTopicsRequest请求创建。
C. 配置broker端参数auto.create.topics.enable为true，然后发送metadataRequest请求。
无论上面哪种方式创建，都是在/brokers/topics下创建对应的znode，然后把topic的分区以及对应副本对应列表都写入znode中。
3）删除topic
A、通过kafka-topics脚本删除--delete。
B、构造deleteTopicsRequest。
这两种方式都是在/brokers/delete_topics下新建znode。
4）分区重分配
分区重分配操作通常是由kafka集群的管理员发起的，旨在对topic所有分区重新分配副本所在broker的位置，以期望实现更均匀分配效果。
5）preferred leader 选举
为了避免分区副本分配不均匀，kafka引入了preferred副本概念，解决分区不均匀问题。
A、设置broker端参数auto.leader.rebalance.enable为true，这样controller定时自动调整preferred leader。
B、通过kafka-preferred-repica-election脚本手动触发。
6）topic分区扩展
在kafka集群运行过程中，用户可能会发现某些topic的现有分区不足以支持业务量，因此需要考虑增加分区。
7）broker加入集群
每个broker成功启动都会在/broker/ids下创建一个znode，并写入broker信息。如果让kafka动态维护broker列表，就必须注册一个zookeeper监听器时刻监控目录下的变化。
8）broker崩溃
由于当前broker在zookeeper中注册znode是临时节点，因此一旦broker崩溃，broker与zookeeper的会话失效并导致临时节点被删除，上面监听broker加入的那些监听器同样用来监视那些推出群聊的broker列表。
9）受控关闭
通过kafka-service-stop是通过脚本、kill -9 、kill -15正常关闭，而broker崩溃或强制推出通常是以“掉电”或者“kill -9”的方式实现的。前者被称为受控关闭。
10）controller leader选举
作为重要组件，controller必定支持故障转移，宕机时候必须保证及时选出新的controller。
1、关闭controller所在broker。
2、当前controller所在broker宕机或者崩溃。
3、手动删除zookeeper下面的/controller节点。
4、说动向/controller下写入新的broker id。
