kafka 集群迁移实践

背景

本文与zookeeper迁移实践 背景类似，属于云平台供应商迁移过程中的中间件迁移。kafka 集群作为公司重要的消息组件，在各业务线中广泛使用，因此，对于 kafka 集群的迁移，也是非常慎重，在参考官方文档及测试集群验证的基础上总结出的迁移步骤，成功应用于线上集群迁移，希望对读者有所帮助。

迁移要求

对于使用方众多的 kafka 集群来说，迁移过程必须要保证数据一致，同时服务不能中断，因为消息队列关乎很多业务方的处理逻辑，比如下单，支付，物流状态等，如果数据不一致或者服务中断，对业务来说是不可接受的。

实施方案

我们目前的 kafka 集群为 kafka_2.11-2.1.1 版本，采用的方案是：

• 先扩展集群到新节点

• 利用 kafka 内置的 partition reassignment 机制将存量数据迁移到新节点

• 下线旧节点

整个操作过程，kafka 集群可以持续对外提供服务。方案图如下：

kafka集群迁移示例

具体迁移流程

1. 启动新节点加入集群(使用相同 zk 即可)

2. 列出所有 topic

bin/kafka-topics.sh --zookeeper localhost:3181 --list

根据得到 topic 列表，拼成规定的 json 格式，e.g. moveTopic.json

{"topics":[{"topic": "foo"},{"topic": "foo1"}],"version":1}

3. 生成迁移计划

bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --zookeeper localhost:2181 --broker-list toBroker1,toBroker2,toBroker3 --generate --topics-to-move-json-file moveTopic.json

其中，toBroker 是要迁往的目标节点编号。将执行结果分别保存两个 json 文件，一个是迁移前的分配(用于有问题时回滚)，另一个是下一步要执行的目标分配。

比如，从第三步的结果我们最终形成 origin.json 和 reassign.json 两个文件

4. 执行 reassign 计划

bin/kafka-reassign-partitions.sh --command-config client.properties --zookeeper localhost:2181 --execute --reassignment-json-file reassign.json --throttle 100000000

其中 --throttle 参数是限制迁移过程中节点的带宽使用，这样可以防止数据迁移对 broker 节点提供正常服务造成影响，比如上述限制带宽为 100MB。

这里有两点需要注意：

A. 如果 kafka 集群开启了验证，则需要

1）export KAFKA_OPTS=" -Djava.security.auth.login.config=${jaasPath}/kafka_server_jaas.conf "

然后再执行上述命令，其中 kafka_server_jaas.conf 是启动 kafka 集群是设置的认证信息文件

2）需要加上 --command-config client.properties ，其中 client.properties 内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

B.上述 kafka-reassign-partitions.sh 命令可重复执行，用于调整 --throttle 参数

5. 检查是否重新分配完毕

使用 --verify 进行判断

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --verify --reassignment-json-file reassign.json

该命令的输出：

xxx in progress 表示 partition 还在重分配中

xxx completed successfully 表示 partition 重分配完成

全部重分配完后会输出(这个要仔细确认) Throttle was removed

6. 重分配完成后，下发新地址串给业务方(建议对外提供负载均衡器地址，broker 地址挂载负载均衡器后面)。

重分配完成后，建议观察一段时间，然后旧 broker 可以安全下线（参考注意事项第三点）。

注意事项

1. 在重启 broker 前确保所有 topic 都至少有 3 个 replica

这是因为在线上环境中，我们通常在 broker 设置 min.insync.replicas=2，而 topic 的副本个数通常也设置为 3，所以如果某些 topic 只有 2 个副本，那么在重启 broker 时，消息发送方可能会出现异常，e.g.

2021-09-07 13:34:52.803 WARN  o.a.k.c.producer.internals.Sender [] [Producer clientId=producer-1] Got error produce response with correlation id 3042 on topic-partition product-sync-jd-pre-0, retrying (0 attempts left). Error: NOT_ENOUGH_REPLICAS

Not Enough Replica 异常

2. 在执行 partition 重分配时，强烈建议加上 --throttle 参数，可以控制带宽使用，特别是公司使用云平台专线的情况，可以保证不影响正常业务方应用运行。

3. 假设原来 kafka 集群为 1,2,3 节点，迁移数据到 4,5,6 节点后，是否可以安全的直接下线 1,2,3 节点而不要求业务方做任何操作？

我们知道，扩展集群到新节点后，如果不尽快下线老的 broker 节点，那么新创建的 topic 还是有可能会将 partition 分配到旧节点上，因此拖得时间越长，旧节点上积攒的数据也就越多。

经过在测试集群做验证，我们发现在业务方链接老的 bootstrap server 节点启动后，kafka 经过扩展集群节点且数据迁移到新节点后，此时直接下线老的节点并不会对业务方有任何影响（短暂连接异常，比如旧节点作为 coordinator 的情况），仍可以正常收发信息。

因此我们得出结论，扩展节点并完成数据迁移后，老节点可以直接下线，这样可以避免进行二次数据迁移。

遇到的问题

• 报错 Caused by: org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.

这是短暂 partition leader 重新选举中的消费或者生产方抛出异常，可以忽略。

• 重分配 partition 后(在新节点)，很多 topic 卡在 under replicated 状态(主副本正常工作)

一开始以为数据量大的缘故(一次性全部 topic 重分配 partition)，网上也有遇到类似的情况，说是 kafka 的某个 bug，重启 zk leader 节点或者 broker 节点可以解决。我试了，重启 zk leader 节点不管用，需要重启每个新 broker 节点才解决。

还有一种情况是，由于我们在 kafka-manager 上查看 topic 状态，所以有时候信息不一定实时，可以重启 kafka-manager 试试。

• kafka 消费组一直报错：Group coordinator kafka.huoli101.com:2093 (id: 2147483641 rack: null) is unavailable or invalid, will attempt rediscovery.

但该节点其实是正常的，telnet 也能连接上。重启该节点，业务方能正常消费，一段时间后又会出现上述异常而停止消费，非常诡异。

根本原因还没有找到...了解到业务方使用了多于 partition 个数的 consumer，以及在出现异常时有超出专线带宽使用限制的情况（如下图所示）。目前采用增加跨网段专线带宽以及减少 consumer 个数来解决，暂时没有继续出现类似异常（从结果看，该异常很可能是短时间带宽占满导致 client 无法连上 broker 导致）。

kafka broker出现很多踢出consumer的日志

专线带宽使用情况

总结

本文记录了在迁移 kafka 集群时采用的解决方案，具体步骤以及遇到的问题和解决方法，并对迁移后老节点是否能直接下线给出了验证结论，希望对大家有所帮助。

大家有类似的迁移经历或者迁移过程中遇到的问题，欢迎留言探讨～

参考文档

kafka官方文档 expanding your cluster