Kafka Controller 是 Kafka 的核心组件，在前面的文章中，已经详细讲述过 Controller 部分的内容。在过去的几年根据大家在生产环境中应用的反馈，Controller 也积累了一些比较大的问题，而针对这些问题的修复，代码的改动量都是非常大的，无疑是一次重构，因此，社区准备在新版的系统里对 Controller 做一些相应的优化（0.11.0及以后的版本），相应的设计方案见：Kafka Controller Redesign，本文的内容就是结合这篇文章做一个简单的总结。

Controller 功能

在一个 Kafka 中，Controller 要处理的事情总结如下表所示：

Controller 目前存在的问题

之所以要重新设计 Controller，是因为现在的 Controller 积累了一些比较难解决的问题，这些问题解决起来，代码改动量都是巨大的，甚至需要改变 controller 部门的设计，基本就跟重构差不多了，下面我们先来了看一下 controller 之前（主要是 0.11.0 之前的版本）存在的一些问题。

目前遇到的比较大的问题有以下几个：

1. Partition 级别同步 zk 写；
2. sequential per-partition controller-to-broker requests；
3. Controller 复杂的并发语义；
4. 代码组织混乱；
5. 控制类请求与数据类请求未分离；
6. Controller 给 broker 的请求中没有 broker 的 generation信息；
7. ZkClient 阻碍 Client 的状态管理。

Partition 级别同步 zk 写

zookeeper 的同步写意味着在下次写之前需要等待前面整个过程的结束，而且由于它们都是 partition 粒度的（一个 Partition 一个 Partition 的去执行写操作），对于 Partition 非常多的集群来说，需要等待的时间会更长，Controller 通常会在下面这两个地方做 Partition 级别 zookeeper 同步写操作：

1. PartitionStateMachine 在进行触发 leader 选举（partition 目的状态是 OnlinePartition），将会触发上面的操作；
2. ReplicaStateMachine 更新 LeaderAndIsr 信息到 zk（replica 状态转变为 OfflineReplica），这种情况也触发这种情况，它既阻碍了 Controller 进程，也有可能会 zk 造成压力。

sequential per-partition controller-to-broker requests

Controller 在向 Broker 发送请求，有些情况下也是 Partition 粒度去发送的，效率非常低，比如在 Controller 处理 broker shutdown 请求时，这里是按 Partition 级别处理，每处理一个 Partition 都会执行 Partition、Replica 状态变化以及 Metadata 更新，并且调用 sendRequestsToBrokers() 向 broker 发送请求，这样的话，效率将变得非常低。

Controller 复杂的并发语义

Controller 需要在多个线程之间共享状态信息，这些线程有：

1. IO threads handling controlled shutdown requests
2. The ZkClient org.I0Itec.zkclient.ZkEventThread processing zookeeper callbacks sequentially；
3. The TopicDeletionManager kafka.controller.DeleteTopicsThread；
4. Per-broker RequestSendThread within ControllerChannelManager.

所有这些线程都需要访问或修改状态信息（ControllerContext），现在它们是通过 ControllerContext 的 controllerLock（排它锁）实现的，Controller 的并发变得虚弱无力。

代码组织混乱

KafkaController 部分的代码组织（KafkaController、PartitionStateMachine 和 ReplicaStateMachine）不是很清晰，比如，下面的问题就很难回答：

1. where and when does zookeeper get updated?
2. where and when does a controller-to-broker request get formed?
3. what impact does a failing zookeeper update or controller-to-broker request have on the cluster state?

这也导致了这部分很多开发者不敢轻易去改动。

控制类请求与数据类请求未分离

现在 broker 收到的请求，有来自 client、broker 和 controller 的请求，这些请求都会被放到同一个 requestQueue 中，它们有着同样的优先级，所以来自 client 的请求很可能会影响来自 controller 请求的处理（如果是 leader 变动的请求，ack 设置的不是 all，这种情况有可能会导致数据丢失）。

Controller 给 broker 的请求中没有 broker 的 generation信息

这里的 Broker generation 代表着一个标识，每当它重新加入集群时，这个标识都会变化。如果 Controller 的请求没有这个信息的话，可能会导致一个重启的 Broker 收到之前的请求，让 Broker 进入到一个错误的状态。

比如，Broker 收到之前的 StopReplica 请求，可能会导致副本同步线程退出。

ZkClient 阻碍 Client 的状态管理

这里的状态管理指的是当 Client 发生重连或会话过期时，Client 可以监控这种状态变化，并做出一些处理，因为开源版的 ZKClient 在处理 notification 时，是线性处理的，一些 notification 会被先放到 ZkEventThread’s queue 中，这样会导致一些最新的 notification 不能及时被处理，特别是与 zk 连接断开重连的情况。

Controller 改进方案

关于上述问题，Kafka 提出了一些改进方案，有些已经在最新版的系统中实现，有的还在规划中。

使用异步的 zk API

Zookeeper 的 client 提供三种执行请求的方式：

1. 同步调用，意味着下次请求需要等待当前当前请求的完成；
2. 异步调用，意味着不需要等待当前请求的完成就可以开始下次请求的执行，并且我们可以通过回调机制去处理请求返回的结果；
3. 单请求的 batch 调用，意味着 batch 内的所有请求都会在一次事务处理中完成，这里需要关注的是 zookeeper 的 server 对单请求的大小是有限制的（jute.maxbuffer）。

文章中给出了三种请求的测试结果，Kafka 最后选取的是异步处理机制，因为对于单请求处理，异步处理更加简洁，并且相比于同步处理还可以保持一个更好的写性能。