Kafka在Yelp的应用十分广泛。事实上，我们每天通过各种集群发送数十亿条消息。在这背后，Kafka使用Zookeeper完成各种分布式协调任务，例如决定哪个Kafka broker负责分配分区首领，以及在broker中存储有关主题的元数据。

Kafka在Yelp的成功应用说明了我们的集群从其首次部署Kafka以来经历了大幅的增长。与此同时，其他的Zookeeper重度用户（例如Smartstack和PaasTA）规模也在增长，给我们的共享Zookeeper集群添加了很多负担。为了缓解这种情况，我们决定让我们的Kafka集群使用专门的Zookeeper集群。

由于我们非常依赖Kafka，因维护造成的任何停机都会导致连锁反应，例如显示给业务所有者的仪表盘出现延迟、日志堆积在服务器上。那么问题就来了：我们是否可以在不引起Kafka及其他Zookeeper用户注意的情况下切换Zookeeper集群？

Zookeeper有丝分裂

经过团队间对Kafka和Zookeeper的几轮讨论和头脑风暴之后，我们找到了一种方法，似乎可以实现我们的目标：在不会导致Kafka停机的情况下让Kafka集群使用专门的Zookeeper集群。

我们提出的方案可以比作自然界的细胞有丝分裂：我们复制Zookeeper主机（即DNA），然后利用防火墙规则（即细胞壁）把复制好的主机分成两个独立的集群。

有丝分裂中的主要事件，染色体在细胞核中分裂

让我们一步一步深入研究细节。在本文中，我们将会用到源集群和目标集群，源集群代表已经存在的集群，目标集群代表Kafka将要迁移到的新集群。我们要用到的示例是一个包含三个节点的Zookeeper集群，但这个过程本身可用于任何数量的节点。

我们的示例将为Zookeeper节点使用以下IP地址：

源 192.168.1.1-3
目标 192.168.1.4-6

第1阶段：DNA复制

首先，我们需要启动一个新的Zookeeper集群。这个目标集群必须是空的，因为在迁移的过程中，目标集群中的内容将被删除。

然后，我们将目标集群中的两个节点和源集群中的三个节点组合在一起，得到一个包含五个节点的Zookeeper集群。这么做的原因是我们希望数据（最初由Kafka保存在源Zookeeper集群中）被复制到目标集群上。Zookeeper的复制机制会自动执行复制过程。

把来自源集群和目标集群的节点组合在一起

每个节点的zoo.cfg文件现在看起来都像下面这样，包含源集群的所有节点和目标集群中的两个节点：

server.1=192.168.1.1:2888:3888
server.2=192.168.1.2:2888:3888
server.3=192.168.1.3:2888:3888
server.4=192.168.1.4:2888:3888
server.5=192.168.1.5:2888:3888

注意，来自目标集群的一个节点（在上面的例子中是192.168.1.6）在该过程中保持休眠状态，没有成为联合集群的一部分，并且Zookeeper也没有在其上运行，这是为了保持源集群的quorum。

此时，联合集群必须重启。确保执行一次滚动重启（每次重启一个节点，期间至少有10秒的时间间隔），从来自目标集群的两个节点开始。这个顺序可以确保源集群的quorum不会丢失，并在新节点加入该集群时确保对其他客户端（如Kafka）的可用性。

Zookeeper节点滚动重启后，Kafka对联合集群中的新节点一无所知，因为它的Zookeeper连接字符串只有原始源集群的IP地址：

zookeeper.connect=192.168.1.1,192.168.1.2,192.168.1.3/kafka

发送给Zookeeper的数据现在被复制到新节点，而Kafka甚至都没有注意到。

现在，源集群和目标集群之间的数据同步了，我们就可以更新Kafka的连接字符串，以指向目标集群：

zookeeper.connect=192.168.1.4,192.168.1.5,192.168.1.6/kafka

需要来一次Kafka滚动重启，以获取新连接，但不要进行整体停机。

第2阶段：有丝分裂

拆分联合集群的第一步是恢复原始源Zookeeper及目标Zookeeper的配置文件（zoo.cfg），因为它们反映了集群所需的最终状态。注意，此时不应重启Zookeeper服务。

我们利用防火墙规则来执行有丝分裂，把我们的联合集群分成不同的源集群和目标集群，每个集群都有自己的首领。在我们的例子中，我们使用iptables来实现这一点，但其实可以两个Zookeeper集群主机之间强制使用的防火墙系统应该都是可以的。

对每个目标节点，我们运行以下命令来添加iptables规则：

$source_node_list = 192.168.1.1,192.168.1.2,192.168.1.3
sudo /sbin/iptables -v -A INPUT  -p tcp -d $source_node_list -j REJECT
sudo /sbin/iptables -v -A OUTPUT  -p tcp -d $source_node_list -j REJECT

这将拒绝从目标节点到源节点的任何传入或传出TCP流量，从而实现两个集群的分隔。

通过防火墙规则分隔源集群和目标集群，然后重启

分隔意味着现在两个目标节点与其他节点是分开的。因为它们认为自己属于一个五节点的集群，而且无法与集群的大多数节点进行通信，所以它们无法进行首领选举。

此时，我们同时重启目标集群中每个节点的Zookeeper，包括那个不属于联合集群的休眠节点。这样Zookeeper进程将使用步骤2中提供的新配置，而且还会强制在目标集群中进行首领选举，从而每个集群都会有自己的首领。

从Kafka的角度来看，目标集群从发生网络分区那一刻起就不可用，直到首领选举结束后才可用。对Kafka来说，这是整个过程中Zookeeper不可用的唯一一个时间段。从现在开始，我们有了两个不同的Zookeeper集群。

现在我们要做的是清理。源集群仍然认为自己还有两个额外的节点，我们需要清理一些防火墙规则。

接下来，我们重启源集群，让只包含原始源集群节点的zoo.cfg配置生效。我们现在可以安全地删除防火墙规则，因为集群之间不再需要相互通信。下面的命令用于删除iptables规则：

$source_node_list = 192.168.1.1,192.168.1.2,192.168.1.3
sudo /sbin/iptables -v -D INPUT  -p tcp -d $source_node_list -j REJECT
sudo /sbin/iptables -v -D OUTPUT  -p tcp -d $source_node_list -j REJECT

树立信心

分布式压力测试

我们用于测试迁移过程正确性的主要方法是分布式压力测试。在迁移过程中，我们通过脚本在多台机器上运行数十个Kafka生产者和消费者实例。当流量生成完成后，所有被消费的数据有效载荷被聚集到单台主机上，以便检测是否发生数据丢失。

分布式压力测试的工作原理是为Kafka生产者和消费者创建一组Docker容器，并在多台主机上并行运行它们。所有生成的消息都包含了一个序列号，可以用于检测是否发生消息丢失。

临时集群

为了证明迁移的正确性，我们需要构建一些专门用于测试的集群。我们不是通过手动创建Kafka集群，然后在测试完以后再关掉它们，而是构建了一个工具，可以在我们的基础架构上自动生成和关闭集群，从而可以通过脚本来执行整个测试过程。

这个工具连接到AWS EC2 API上，并用特定的EC2实例标签激活多台主机，允许我们的puppet代码配置主机和安装Kafka（通过External Node Classifiers，https://puppet.com/docs/puppet/5.5/nodes_external.html）。这样我们就可以重新运行迁移脚本，并多次模拟迁移过程。

这个临时集群脚本后来被用于创建临时Elasticsearch集群进行集成测试，这证明了它是一个非常有用的工具。

zk-smoketest

我们发现，phunt的Zookeeper smoketest脚本（https://github.com/phunt/zk-smoketest）在迁移过程中可用于监控每个Zookeeper集群的状态。在迁移的每个阶段，我们在后台运行smoketest，以确保Zookeeper集群的行为符合预期。

zkcopy

我们的第一个用于迁移的计划涉及关闭Kafka、把Zookeeper数据子集复制到新集群、使用更新过的Zookeeper连接重启Kafka。迁移过程的一个更精细的版本——我们称之为“阻止和复制（block & copy）”——被用于把Zookeeper客户端迁移到存有数据的集群，这是因为“有丝分裂”过程需要一个空白的目标Zookeeper集群。用于复制Zookeeper数据子集的工具是zkcopy（https://github.com/ksprojects/zkcopy），它可以把Zookeeper集群的子树复制到另一个集群中。

我们还添加了事务支持，让我们可以批量管理Zookeeper操作，并最大限度地减少为每个znode创建事务的网络开销。这使我们使用zkcopy的速度提高了约10倍。

另一个加速迁移过程的核心功能是“mtime”支持，它允许我们跳过复制早于给定修改时间的节点。我们因此避免了让Zookeeper集群保持同步的第2个“catch-up”复制所需的大部分工作。Zookeeper的停机时间从25分钟减少为不到2分钟。

经验教训

Zookeeper集群是轻量级的，如果有可能，尽量不要在不同服务之间共享它们，因为它们可能会引起Zookeeper的性能问题，这些问题很难调试，并且通常需要停机进行修复。

我们可以在Kafka不停机的情况下让Kafka使用新的Zookeeper集群，但是，这肯定不是一件小事。

如果在进行Zookeeper迁移时允许Kafka停机，那就简单多了。

阅读英文原文：Migrating Kafka’s Zookeeper With No Downtime；https://engineeringblog.yelp.com/2019/01/migrating-kafkas-zookeeper-with-no-downtime.html

创作场景

不停止 Kafka，如何将旧集群迁移到专有的 Zookeeper 集群？