Kafka的Controller Broker

[TOC]
控制器组件（Controller），是 Apache Kafka 的核心组件。它的主要作用是在 Apache ZooKeeper 的帮助下管理和协调整个 Kafka 集群。集群中任意一台 Broker 都能充当控制器的角色，但是，在运行过程中，只能有一个 Broker 成为控制器，行使其管理和协调的职责。接下来，我们将讨论Controller原理和内部运行机制。

什么是Controller Broker

在分布式系统中，通常需要有一个协调者，该协调者会在分布式系统发生异常时发挥特殊的作用。在Kafka中该协调者称之为控制器(Controller),其实该控制器并没有什么特殊之处，它本身也是一个普通的Broker，只不过需要负责一些额外的工作(追踪集群中的其他Broker，并在合适的时候处理新加入的和失败的Broker节点、Rebalance分区、分配新的leader分区等)。值得注意的是：Kafka集群中始终只有一个Controller Broker。

控制器（Controller）是Kafka的核心组件，主要作用是在ZK的帮助下管理和协调整个Kafka集群
集群中任一Broker都能充当控制器的角色，但在运行过程中，只能有一个Broker成为控制器，行使管理和协调的职责

[zk: localhost:2181(CONNECTED) 1] get /controller
{"version":1,"brokerid":0,"timestamp":"1571311742367"}
cZxid = 0xd68
ctime = Thu Oct 17 19:29:02 CST 2019
mZxid = 0xd68
mtime = Thu Oct 17 19:29:02 CST 2019
pZxid = 0xd68
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1000209974b0000
dataLength = 54
numChildren = 0

zookeeper

Kafka控制器重度依赖ZK
ZK是一个提供高可靠性的分布式协调服务框架
ZK使用类似于文件系统的树形结构，根目录以/开始，结构上的每个节点称为znode，用来保存一些元数据协调信息
如果以znode的持久性来划分，znode可以分为持久性znode和临时znode
持久性znode不会因为ZK集群重启而消失
临时znode则会与创建该znode的ZK会话绑定，一旦会话结束，该节点会被自动删除
ZK赋予客户端监控znode变更的能力，即所谓的Watch通知功能
一旦znode节点被创建、删除、子节点数量发生变化，znode所存的数据本身发生变更
ZK会通过节点变更监听器（ChangeHandler）的方式显式通知客户端
ZK被用来实现集群成员管理、分布式锁、领导者选举等功能，Kafka控制器大量使用Watch功能实现对集群的协调管理

/controller节点

Broker在启动时，会尝试去ZK创建/controller节点
第一个成功创建/controller节点的Broker会被指定为为控制器

控制器的职责

主题管理
1. 完成对Kafka主题的创建、删除以及分区增加的操作
2. 执行kafka-topics时，大部分的后台工作都是由控制器完成的
分区重分配
分区重分配主要是指kafka-reassign-partitions脚本提供的对已有主题分区进行细粒度的分配功能
Preferred领导者选举
Preferred领导者选举主要是Kafka为了避免部分Broker负载过重而提供的一种换Leader的方案
集群成员管理
- 自动检测新增Broker、Broker主动关闭、Broker宕机
- 自动检测依赖于Watch功能和ZK临时节点组合实现的
  1. 控制器会利用Watch机制检查ZK的/brokers/ids节点下的子节点数量变更
  2. 当有新Broker启动后，它会在/brokers/ids/下创建专属的临时znode节点
  3. 一旦创建完毕，ZK会通过Watch机制将消息通知推送给控制器，控制器能够自动感知这个变化当Broker宕机或者主动关闭后，该Broker与ZK的会话结束，这个znode会被自动删除
  4. 当Broker宕机或者主动关闭后，该Broker与ZK的会话结束，这个znode会被自动删除, ZK的Watch机制会将这一变更推送给控制器
数据服务

向其它Broker提供数据服务，控制器上保存了最全的集群元数据信息
其它Broker会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据

Controller Broker是如何被选出来的

上一小节解释了什么是Controller Broker，并且每台 Broker 都有充当控制器的可能性。那么，控制器是如何被选出来的呢？当集群启动后，Kafka 怎么确认控制器位于哪台 Broker 呢？

实际上，Broker 在启动时，会尝试去 ZooKeeper 中创建 /controller 节点。Kafka 当前选举控制器的规则是：第一个成功创建 /controller 节点的 Broker 会被指定为控制器。

Controller Broker的具体作用是什么

Controller Broker的主要职责有很多，主要是一些管理行为，主要包括以下几个方面：

创建、删除主题，增加分区并分配leader分区
集群Broker管理（新增 Broker、Broker 主动关闭、Broker 故障)
preferred leader选举
分区重分配

处理集群中下线的Broker

当某个Broker节点由于故障离开Kafka群集时，则存在于该Broker的leader分区将不可用(由于客户端仅对leader分区进行读写操作)。为了最大程度地减少停机时间，需要快速找到替代的leader分区。

Controller Broker可以对失败的Broker做出响应，Controller Broker可以从Zookeeper监听(zookeeper watch)中获取通知信息，ZooKeeper 赋予客户端监控 znode 变更的能力，即所谓的 Watch 通知功能。一旦 znode 节点被创建、删除，子节点数量发生变化，抑或是 znode 所存的数据本身变更，ZooKeeper 会通过节点变更监听器 (ChangeHandler) 的方式显式通知客户端。

每个 Broker 启动后，会在zookeeper的 /Brokers/ids 下创建一个临时 znode。当 Broker 宕机或主动关闭后，该 Broker 与 ZooKeeper 的会话结束，这个 znode 会被自动删除。同理，ZooKeeper 的 Watch 机制将这一变更推送给控制器，这样控制器就能知道有 Broker 关闭或宕机了，从而进行后续的协调操作。

Controller将收到通知并对此采取行动，决定哪些Broker上的分区成为leader分区，然后，它会通知每个相关的Broker，要么将Broker上的主题分区变成leader，要么通过LeaderAndIsr请求从新的leader分区中复制数据。

处理新加入到集群中的Broker

通过将Leader分区副本均匀地分布在集群的不同Broker上，可以保障集群的负载均衡。在Broker发生故障时，某些Broker上的分区副本会被选举为leader，会造成一个Broker上存在多个leader分区副本的情况，由于客户端只与leader分区副本交互，所以这会给Broker增加额外的负担，并损害集群的性能和运行状况。因此，尽快恢复平衡对集群的健康运行是有益的。

Kafka认为leader分区副本最初的分配（每个节点都处于活跃状态）是均衡的。这些被最初选中的分区副本就是所谓的首选领导者(preferred leaders)。由于Kafka还支持机架感知的leader选举(rack-aware leader election) ,即尝试将leader分区和follower分区放置在不同的机架上，以增加对机架故障的容错能力。因此，leader分区副本的存在位置会对集群的可靠性产生影响。

默认情况下auto.leader.rebalance.enabled为true，表示允许 Kafka 定期地对一些 Topic 分区进行
Leader 重选举。大部分情况下，Broker的失败很短暂，这意味着Broker通常会在短时间内恢复。所以当节点离开群集时，与其相关联的元数据并不会被立即删除。

当Controller注意到Broker已加入集群时，它将使用Broker ID来检查该Broker上是否存在分区，如果存在，则Controller通知新加入的Broker和现有的Broker，新的Broker上面的follower分区再次开始复制现有leader分区的消息。为了保证负载均衡，Controller会将新加入的Broker上的follower分区选举为leader分区。

注意：上面提到的选Leader分区，严格意义上是换Leader分区，为了达到负载均衡，可能会造成原来正常的Leader分区被强行变为follower分区。换一次 Leader 代价是很高的，原本向 Leader分区A(原Leader分区) 发送请求的所有客户端都要切换成向 B (新的Leader分区)发送请求，建议你在生产环境中把这个参数设置成 false。

同步副本(in-sync replica ,ISR)列表

SR中的副本都是与Leader进行同步的副本，所以不在该列表的follower会被认为与Leader是不同步的. 那么，ISR中存在是什么副本呢？首先可以明确的是：Leader副本总是存在于ISR中。而follower副本是否在ISR中，取决于该follower副本是否与Leader副本保持了“同步”。

始终保证拥有足够数量的同步副本是非常重要的。要将follower提升为Leader，它必须存在于同步副本列表中。每个分区都有一个同步副本列表，该列表由Leader分区和Controller进行更新。

选择一个同步副本列表中的分区作为leader 分区的过程称为clean leader election。注意，这里要与在非同步副本中选一个分区作为leader分区的过程区分开，在非同步副本中选一个分区作为leader的过程称之为unclean leader election。由于ISR是动态调整的，所以会存在ISR列表为空的情况，通常来说，非同步副本落后 Leader 太多，因此，如果选择这些副本作为新 Leader，就可能出现数据的丢失。毕竟，这些副本中保存的消息远远落后于老 Leader 中的消息。在 Kafka 中，选举这种副本的过程可以通过Broker 端参数 *unclean.leader.election.enable *控制是否允许 Unclean 领导者选举。开启 Unclean 领导者选举可能会造成数据丢失，但好处是，它使得分区 Leader 副本一直存在，不至于停止对外提供服务，因此提升了高可用性。反之，禁止 Unclean Leader 选举的好处在于维护了数据的一致性，避免了消息丢失，但牺牲了高可用性。分布式系统的CAP理论说的就是这种情况。

脑裂

如果controller Broker 挂掉了，Kafka集群必须找到可以替代的controller，集群将不能正常运转。这里面存在一个问题，很难确定Broker是挂掉了，还是仅仅只是短暂性的故障。但是，集群为了正常运转，必须选出新的controller。如果之前被取代的controller又正常了，他并不知道自己已经被取代了，那么此时集群中会出现两台controller。

其实这种情况是很容易发生。比如，某个controller由于GC而被认为已经挂掉，并选择了一个新的controller。在GC的情况下，在最初的controller眼中，并没有改变任何东西，该Broker甚至不知道它已经暂停了。因此，它将继续充当当前controller，这是分布式系统中的常见情况，称为脑裂。

假如，处于活跃状态的controller进入了长时间的GC暂停。它的ZooKeeper会话过期了，之前注册的/controller节点被删除。集群中其他Broker会收到zookeeper的这一通知。

image.png

由于集群中必须存在一个controller Broker，所以现在每个Broker都试图尝试成为新的controller。假设Broker 2速度比较快，成为了最新的controller Broker。此时，每个Broker会收到Broker2成为新的controller的通知，由于Broker3正在进行”stop the world”的GC，可能不会收到Broker2成为最新的controller的通知。

image.png

等到Broker3的GC完成之后，仍会认为自己是集群的controller，在Broker3的眼中好像什么都没有发生一样。

image.png

现在，集群中出现了两个controller，它们可能一起发出具有冲突的命令，就会出现脑裂的现象。如果对这种情况不加以处理，可能会导致严重的不一致。所以需要一种方法来区分谁是集群当前最新的Controller。

Kafka是通过使用epoch number（纪元编号，也称为隔离令牌）来完成的。epoch number只是单调递增的数字，第一次选出Controller时，epoch number值为1，如果再次选出新的Controller，则epoch number将为2，依次单调递增。

每个新选出的controller通过Zookeeper 的条件递增操作获得一个全新的、数值更大的epoch number 。其他Broker 在知道当前epoch number 后，如果收到由controller发出的包含较旧(较小)epoch number的消息，就会忽略它们，即Broker根据最大的epoch number来区分当前最新的controller。

image.png

上图，Broker3向Broker1发出命令:让Broker1上的某个分区副本成为leader，该消息的epoch number值为1。于此同时，Broker2也向Broker1发送了相同的命令，不同的是，该消息的epoch number值为2，此时Broker1只听从Broker2的命令(由于其epoch number较大)，会忽略Broker3的命令，从而避免脑裂的发生。

控制器内部设计原理

image.png

在Kafka 0.11之前，控制器的设计是相当繁琐的，导致很多Bug无法修复
控制器是多线程的设计，会在内部创建很多个线程
1. 控制器需要为每个Broker都创建一个对应的Socket连接，然后再创建一个专属的线程，用于向这些Broker发送请求
2. 控制器连接ZK的会话，也会创建单独的线程来处理Watch机制的通知回调
3. 控制器还会为主题删除创建额外的IO线程
这些线程还会访问共享的控制器缓存数据，多线程访问共享可变数据是维持线程安全的最大难题，为了保护数据安全性，控制器在代码中大量使用ReentrantLock同步机制，进一步拖慢整个控制器的处理速度
社区在0.11版本重构了控制器的底层设计，把多线程的方案改成了单线程+事件队列的方案
- 引进了事件处理器，统一处理各种控制器事件
- 控制器将原来执行的操作全部建模成独立的事件，发送到专属的事件队列中，供事件处理器消费
- 单线程：控制器只是把缓存状态变更方面的工作委托给了这个线程而已
- 优点：控制器缓存中保存的状态只被一个线程处理，因此不需要重量级的线程同步机制来维护线程安全
针对控制器的第二个改进：将之前同步操作ZK全部换成异步操作
- ZK本身的API提供了同步写和异步写两种方式
- 之前控制器操作ZK时使用的是同步API，性能很差
- 当有大量主题分区发生变更时，ZK容易成为系统的瓶颈
Kafka从2.2开始，将控制器发送的请求和普通数据类的请求分开，实现控制器请求单独处理的逻辑
- 之前Broker对接收到的所有请求都一视同仁，不会区别对待
- 如果删除了某个主题，那么控制器会给主题的所有副本所在的Broker发送StopReplica请求
- 如果此时Broker上有大量Produce请求堆积，那么StopReplica请求只能排队
- 既然主题都要被删除了，继续处理Produce请求就显得很没有意义

小结

Kafka Controller，它其实就是一个普通的Broker，除了需要负责一些额外的工作之外，其角色与其他的Broker基本一样。另外还介绍了Kafka Controller的主要职责，并对其中的一些职责进行了详细解释，最后还说明了kafka是如何避免脑裂的。