Controller

控制器组件(Controller)是 Apache Kafka 的核心组件。它的主要作用是在 Apache ZooKeeper 的帮助下管理和协调整个 Kafka 集群，例如分区领导者副本的选举。

集群中任意一台 Broker 都能充当控制器的角色，但是，在运行过程中，只能有一个 Broker 成为控制器，行使其管理和协调的职责。换句话说，每个正常运转的 Kafka 集群，在任意时刻都有且只有一个控制器。

对ZK的依赖

Kafka 控制器大量使用 Watch 功能实现对集群的协调管理。
持久化一部分数据,实现数据的存储功能

下面这张图显示了ZK 在kafka 集群中的作用

数据存储

主要存储了下面的数据，虽然对外提供数据服务的实Controller，但是Controller初始化的时候数据是来自ZK的，也就是说集群 Broker 是不会与 ZooKeeper 直接交互去获取元数据的。相反地，它们总是与 Controller 进行通信，获取和更新最新的集群数据。

故障转移

在 Kafka 集群运行过程中，只能有一台 Broker 充当控制器的角色，那么这就存在单点失效（Single Point of Failure）的风险
故障转移指的是，当运行中的控制器突然宕机或意外终止时，Kafka 能够快速地感知到，并立即启用备用控制器来代替之前失败的控制器。这个过程就被称为 Failover，该过程是自动完成的，无需你手动干预

开始的时候，Broker 0 是控制器。当 Broker 0 宕机后，ZooKeeper 通过 Watch 机制感知到并删除了 /controller 临时节点。之后，所有存活的 Broker 开始竞选新的控制器身份。Broker 3 最终赢得了选举，成功地在 ZooKeeper 上重建了 /controller 节点。之后，Broker 3 会从 ZooKeeper 中读取集群元数据信息，并初始化到自己的缓存中。至此，控制器的 Failover 完成，可以行使正常的工作职责了

Self-Managed Metadata Quorum

目前，Apache Kafka 使用 Apache ZooKeeper 来存储它的元数据，比如分区的位置和主题的配置等数据就是存储在 ZooKeeper 集群中。在 2019 年社区提出了一个计划，以打破这种依赖关系，并将元数据管理引入 Kafka 本身。其实这和早期的offset 信息存储在ZK 上有点像，后来社区采用了内部主题的方式来存储offset 信息，主要是zk 不适合频繁更新和存储大量数据。

目前的设计主要的问题是我们使用kakfa 的时候其实是要安装部署，以及运维管理两套组件集群的，除了kafka 本身，还有就是ZK集群，还有就是外部存储元数据限制了 Kafka 的可伸缩性。当 Kafka 集群启动时，或者一个新的控制器被选中时，控制器必须从 ZooKeeper 加载集群的完整状态。随着元数据数量的增加，加载过程需要的时间也会增加，这限制了 Kafka 可以存储的分区数量

KIP-500 概述了在 Kafka 中处理元数据的更好方法。我们可以将其称为“ Kafka on Kafka”，因为它涉及将 Kafka 的元数据存储在 Kafka 本身中，而不是存储在 ZooKeeper 之类的外部系统中。

在实现了 KIP-500 的系统，元数据将存储在 Kafka 内的分区中，而不是存储在 ZooKeeper。控制器将成为该分区的 leader；除了 Kafka 本身，不需要配置和管理外部元数据系统。

我们将元数据视为日志。Brokers 如果需要最新的数据只能读取日志的末尾。这类似于需要最新日志条目的使用者仅需要读取日志的最后而不是整个日志的方式。Brokers 还可以在进程重新启动时持久化它们的元数据缓存。

控制器体系结构

Kafka 集群选择一个控制器节点来管理分区 leaders 和集群的元数据。我们拥有的分区和元数据越多，控制器的可伸缩性就越重要。我们希望最小化与主题或分区数量成线性比例的时间的操作数量。

其中一个操作是控制器故障转移。当前，当 Kafka 选择新控制器时，它需要在继续之前加载整个集群的状态。随着集群元数据数量的增长，这个过程会花费越来越长的时间。

相比之下，在实现了 KIP-500 的系统，将会有几个备用控制器，随时准备在活动控制器挂掉时接管。这些备用控制器只是元数据分区的 Raft quorum 中的其他节点。这种设计确保了当选择一个新的控制器时，我们永远不需要经过漫长的加载过程。

KIP-500 将加快主题的创建和删除。当前，创建或删除主题时，控制器必须从 ZooKeeper 重新加载集群中所有主题名称的完整列表。这是必要的，因为当 ZooKeeper 通知我们集群中的主题集发生更改时，它不会确切地告诉我们添加或删除了哪些主题。相反，在完全实现了 KIP-500 之后，创建或删除主题只需要在元数据分区中创建一个新条目，这是 O(1）操作。

元数据可伸缩性是将来扩展 Kafka 的关键部分。我们期望单个 Kafka 集群最终将能够支持一百万个分区或更多而且还能够简化 Kafka 的部署和配置

Controller 的选举

Broker 在启动时，会尝试去 ZooKeeper 中创建 /controller 节点。Kafka 当前选举控制器的规则是：第一个成功创建 /controller 节点的 Broker 会被指定为控制器
每个broker启动的时候会去尝试去读取/controller节点的brokerid的值，如果读取到brokerid的值不为-1，则表示已经有其它broker节点成功竞选为控制器，所以当前broker就会放弃竞选
如果Zookeeper中不存在/controller这个节点，或者这个节点中的数据异常，那么就会尝试去创建/controller这个节点，当前broker去创建节点的时候，也有可能其他broker同时去尝试创建这个节点，只有创建成功的那个broker才会成为控制器，而创建失败的broker则表示竞选失败。
每个broker都会在内存中保存当前控制器的brokerid值，这个值可以标识为activeControllerId。

controller_epoch

Zookeeper中还有一个与控制器有关的/controller_epoch节点，这个节点是持久（PERSISTENT）节点，节点中存放的是一个整型的controller_epoch值。
controller_epoch用于记录控制器发生变更的次数，即记录当前的控制器是第几代控制器，我们也可以称之为“控制器的纪元”
controller_epoch的初始值为1，即集群中第一个控制器的纪元为1，当控制器发生变更时，每选出一个新的控制器就将该字段值加1。每个和控制器交互的请求都会携带上controller_epoch这个字段，如果请求的controller_epoch值小于内存中的controller_epoch值，则认为这个请求是向已经过期的控制器所发送的请求，那么这个请求会被认定为无效的请求。
如果请求的controller_epoch值大于内存中的controller_epoch值，那么则说明已经有新的控制器当选了。由此可见，Kafka通过controller_epoch来保证控制器的唯一性，进而保证相关操作的一致性。

控制器的主要作用

主题管理(创建、删除、增加分区)

这里的主题管理，就是指控制器帮助我们完成对 Kafka 主题的创建、删除以及分区增加的操作。换句话说，当我们执行 kafka-topics 脚本时，大部分的后台工作都是控制器来完成的。

分区重分配

分区重分配主要是指，kafka-reassign-partitions 脚本，提供的对已有主题分区进行细粒度的分配功能。这部分功能也是控制器实现的。

Preferred 领导者选举

Preferred 领导者选举主要是 Kafka 为了避免部分 Broker 负载过重而提供的一种换 Leader 的方案

集群成员管理(新增 Broker、Broker 主动关闭、Broker 宕机)

包括自动检测新增 Broker、Broker 主动关闭及被动宕机。这种自动检测是依赖于前面提到的 Watch 功能和 ZooKeeper 临时节点组合实现的。
比如，控制器组件会利用 Watch 机制检查 ZooKeeper 的 /brokers/ids 节点下的子节点数量变更。目前，当有新 Broker 启动后，它会在 /brokers 下创建专属的 znode 节点。一旦创建完毕，ZooKeeper 会通过 Watch 机制将消息通知推送给控制器，这样，控制器就能自动地感知到这个变化，进而开启后续的新增 Broker 作业。
侦测 Broker 存活性则是依赖于刚刚提到的另一个机制：临时节点。每个 Broker 启动后，会在 /brokers/ids 下创建一个临时 znode。当 Broker 宕机或主动关闭后，该 Broker 与 ZooKeeper 的会话结束，这个 znode 会被自动删除。同理，ZooKeeper 的 Watch 机制将这一变更推送给控制器，这样控制器就能知道有 Broker 关闭或宕机了，从而进行“善后”。

数据服务

就是向其他 Broker 提供数据服务。控制器上保存了最全的集群元数据信息，其他所有 Broker 会定期接收控制器发来的元数据更新请求，从而更新其内存中的缓存数据。主要包下面的数据

所有主题信息。包括具体的分区信息，比如领导者副本是谁，ISR 集合中有哪些副本等。
所有 Broker 信息。包括当前都有哪些运行中的 Broker，哪些正在关闭中的 Broker 等。
所有涉及运维任务的分区。包括当前正在进行 Preferred 领导者选举以及分区重分配的分区列表
其实在 ZooKeeper 中也保存了一份,每当控制器初始化时，它都会从 ZooKeeper 上读取对应的元数据并填充到自己的缓存中。有了这些数据，控制器就能对外提供数据服务了。这里的对外主要是指对其他 Broker 而言，控制器通过向这些 Broker 发送请求的方式将这些数据同步到其他 Broker 上

kafka系列之Controller(13)