如何在不出现延迟反应的情况下迁移Kafka（以Yelp为例）

kafka与zookeeper之间的关系

Apache Kafka是分布式发布-订阅消息系统，它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。

kafka对消息保存时根据主题进行归类，发送消息者成为Producer，消息接受者成为Consumer，此外kafka集群有多个kafka实例组成，每个实例(server)成为代理（broker）。无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些元数据。

Yelp（美国最大点评网站）就广泛使用了Kafka。事实上，他们每天通过各种集群发送数十亿条消息。在后台，Kafka使用Zookeeper进行各种分布式协调任务，例如决定由哪个Kafka代理负责分配分区负责人并在其代理中存储有关主题的元数据。

Kafka在Yelp中的成功应用也意味着Yelp的集群从首次部署时就开始大幅增长，与此同时，他们的其他重量级Zookeeper用户（例如，Smartstack和PaasTA）的规模也在增加，从而给Yelp共享的Zookeeper集群带来了更多的载荷。为了缓解这种情况，Yelp决定迁移他们的Kafka集群以使用专用的ZooKeeper集群。

由于Yelp非常依赖Kafka，所以由于维护而造成的任何停机都会引起连锁效应，例如Yelp向业务所有者展示的仪表板会出现延迟反应，以及备份在Yelp服务器上的日志也会消失。所以Yelp的难点就在于如何可以在没有Kafka和其他ZooKeeper用户注意的情况下切换Zookeeper集群。

Zookeeper有丝分裂（Mitosis）

经过几轮讨论并在负责Kafka和Zookeeper的团队之间进行头脑风暴之后，Yelp找到了一种似乎可以让Yelp实现目标的方法：将Kafka集群迁移到他们自己的专用Zookeeper集群中，这样Kafka就不需要停机了。

这个过程与细胞有丝分裂的过程及其相似：Yelp复制Zookeeper宿主(Yelp的DNA)，然后使用防火墙规则(Yelp的细胞壁)将复制的宿主分成两个单独的集群。

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

在本文中，Yelp将引用源集群和目标集群，其中源集群表示已经存在的集群，而目标集群则是Kafka将要迁移到的新集群。Yelp将使用的示例是一个3节点的Zookeeper集群，但是这个过程本身可以用于任意数量的节点。

Yelp的示例将使用Zookeeper节点的以下IP地址：

· 来源地址：192.168.1.1-3

·目标地址：192.168.1.4-6

第一阶段: 复制Zookeeper宿主

首先，Yelp需要启动一个新的Zookeeper集群。此目标群集必须完全为空，因为在迁移过程中内容将被清除。

然后，Yelp将获取两个目标节点并将它们添加到源集群中，从而得到一个5节点的Zookeeper集群。这样做的原因是，Yelp希望将数据(最初由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

注意，来自目标ZooKeeper集群的一个节点(在上面的示例中是192.168.1.6)在此过程中保持休眠状态，它并没有成为联合集群的一部分，ZooKeeper并没有在它上面运行。休眠的原因是为了保持源ZooKeeper集群中的Quorum机制（一种权衡机制）。

此时，必须重新启动关联的ZooKeeper集群。这样做是为了确保从目标集群中的两个节点开始执行滚动重新启动(是针对多个服务器而言的,就是一个一个按顺序的重启所有的服务器,不是同是重启所有的)。以本示例来说，每次重新启动一个节点，每个节点之间的间隔至少为10秒。这个顺序可以确保在源ZooKeeper集群中不会丢失Quorum机制，并确保在新节点加入集群时，其他客户端(例如Kafka)的可用性。

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

zookeeper.connect=192.168.1.1,192.168.1.2,192.168.1.3/kafka

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

现在源集群和目标集群之间的数据是同步的，Yelp可以更新Kafka的Zookeeper连接字符串来指向目标集群。

zookeeper.connect=192.168.1.4,192.168.1.5,192.168.1.6/kafka

需要滚动重启Kafka才能获取新连接，但这并不需要集群整体停机。

第二阶段：有丝分裂

拆分联合集群的第一步是恢复原始源和目标ZooKeeper配置文件(zoo.cfg)，因为它们反映了集群所需的最终状态。请注意，此时不应该重新启动任何Zookeeper服务。

Yelp会使用防火墙规则来执行有丝分裂，将Yelp的联合ZooKeeper集群分成不同的源集群和目标集群，每个集群都有自己的leader。在Yelp的示例中，Yelp使用 iptables 来实现这一点，但这足以让你在两个Zookeeper集群中的主机之间，执行任何防火墙系统。

译者注：IPTABLES 是与最新的 3.5 版本 Linux 内核集成的 IP 信息包过滤系统。如果 Linux 系统连接到因特网或 LAN、服务器或连接 LAN 和因特网的代理服务器， 则该系统有利于在 Linux 系统上更好地控制 IP 信息包过滤和防火墙配置。

对于每个目标节点，Yelp可以运行以下命令来添加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流量，从而实现两个群集的分离。

源集群和目标集群由防火墙规则分隔，然后重新启动

拆分意味着Yelp现在有两个独立的目标节点，由于他们认为自己是5节点集群的一部分，无法与大多数群集通信，因此他们不会选择leader。

此时，Yelp同时在目标集群中的每个节点上重新启动Zookeeper，包括不属于联合集群的休眠节点。这允许Zookeeper进程从第二步中获取其新配置，它还强制在目标群集中选择leader，以便每个群集都有自己的leader。

从Kafka的角度来看，在添加网络分区到leader选择完成之间，目标群集都不可用。这是整个过程中Kafka唯一无法使用ZooKeeper的时间。从现在开始，Yelp有两个不同的Zookeeper集群。此时，在集群之间复制数据就不会出现Kafka数据丢失或停机的情况。

Yelp现在要做的就是把自己清理干净，因为此时源集群仍然认为它有两个额外的节点，所以Yelp需要清除防火墙规则。

接下来，Yelp重新启动源集群，以获取只包含原始源集群节点的zoo.cfg配置。这允许Yelp安全地删除防火墙规则，因为集群不再试图彼此通信。删除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

结果测试

分布式压力测试

测试Yelp迁移过程是否正确的主要方法是分布式应力测试，在迁移过程正在进行时，该脚本会在多台计算机上运行数十个Kafka producer和consumer实例。在流量生成完成后，所有消耗的有效载荷都将被聚合到单个主机中，以检查数据丢失情况。

分布式压力测试的工作原理是为Kafka 的producer和consumer创建一组 Docker 容器，并在多个主机上并行运行它们，其中的列表作为实验参数之一被传送。所有生成的消息都包含可用于检查消息丢失的序列号。

临时的集群

为了证明这个迁移过程是可行的，Yelp希望构建一些专门用于测试的集群。Yelp没有手动构建Kafka集群并再次将其拆分，而是构建了一个工具，用于在Yelp的基础架构中启动可以自动拆除的新集群，从而允许Yelp编写整个测试过程的脚本。

该 工具 使用的是AWS EC2 API，并使用特定的EC2实例标记启动多个主机，从而允许Yelp的puppet代码通过外部 节点分类器 了解如何配置主机和安装Kafka。这最终使Yelp能够运行并重新运行Yelp的迁移脚本，多次模拟迁移。

Ephemeral Cluster脚本后来被重新用于为Yelp的集成测试创建短暂的Elasticsearch集群，这证明它是一个非常宝贵的工具。

ElasticSearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用 Java 开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。

zk-smoketest

Yelp发现phunt的简单 Zookeeper smoketest 脚本在Yelp迁移时，会监控每个Zookeeper集群的状态。在迁移的每个阶段，Yelp都在后台进行Smoke Test (冒烟测试) ，以确保Zookeeper集群的行为符合预期。

zkcopy

Yelp的第一个迁移计划只涉及停止Kafka，将Zookeeper数据的子集复制到新的集群，并使用更新的Zookeeper连接启动Kafka。这个过程的一个更完善的版本，Yelp称为“block & copy”，仍然用于将Zookeeper客户端移动到存储有数据的集群，因为“有丝分裂”过程需要一个空白的目标Zookeeper集群。复制Zookeeper数据子集的一个很好的工具是zkcopy，它会将一个Zookeeper集群的子树复制到另一个集群。

Yelp还添加了多项操作支持，使Yelp能够批量处理Zookeeper操作，并将每个znode创建一个事务的网络运行消耗量降到最低，这使Yelp使用zkcopy的速度提高了10倍!

加速的另一个核心特性是'mtime'支持，它允许Yelp跳过复制任何超过给定修改时间的节点。通过这种方式，Yelp可以避免为使Zookeeper集群同步而产生的复制过程。这样，Zookeeper所需的停机时间从25分钟变为不到2分钟！

总结

Zookeeper集群非常轻量级，如果可能，尽量不要在不同的服务之间共享它们，因为它们可能会在Zookeeper中发生性能变化，且很难调试，通常需要停机来修复。

要将Kafka迁移到一个新的Zookeeper集群，而不需要Kafka停机，但这肯定很难办到。当然，如果你可以安排足够的Kafka停机时间来执行你的Zookeeper迁移，那么将会简单得多。