消息队列MQ
AutoMQ 系统测试体系揭秘
01 前言
Apache Kafka 有着比较完备的自测体系。除了常规的单元测试和集成测试以外，Apache Kafka 还有着 1000+ 的“系统集成和性能测试”（以下简称系统测试）。系统测试会拉起一套真实的 Kafka 集群，并模拟用户使用 Kafka 集群的方式，进行消息的收发，并验证测试结果是否符合预期。整个流程可以实现自动化运行。
AutoMQ 继承了 Apache Kafka 的系统测试框架，并利用该框架保障代码的健壮性，以及对 Apache Kafka 的 100% 兼容性。本文将简要介绍系统测试的原理、演示系统测试过程，并给出 AutoMQ 对系统测试的实践。
02 系统测试框架介绍
2.1 基本信息
系统测试基于 ducktape[1] 框架，该框架旨在让开发者可以像使用单元测试一样编写系统测试的 case。ducktape 框架并不仅仅支持 Kafka 的系统测试。理论上，任意分布式系统都可以基于 ducktape 框架，搭建自己的系统测试体系。
Ducktape 提供如下特性：
ꔷ 保障测试隔离；
ꔷ 多种环境下（Docker、k8s、云厂商 VM）Service 的拉起和销毁；
ꔷ 自定义事件的触发（例如 Kafka node 的强制关机）；
ꔷ 测试结果的收集和总结；
Apache Kafka 目前支持以下三种集成测试环境：
ꔷ Docker;
ꔷ 虚拟机;
ꔷ AWS EC2;
AutoMQ 主要使用的是 Docker 环境下的系统测试。在后续的内容中，我们默认系统测试的使用环境为 Docker，默认工程为 AutoMQ[2] 而非 Apache Kafka。
2.2 目录结构
系统测试的基本目录为工程主目录下的 tests 子目录。
主要的子目录包括：
ꔷ Docker：系统测试运行在 Docker 环境时，镜像的打包逻辑、测试的触发脚本等；
ꔷ kafkatest：系统测试的主目录，包含 Kafka 对 ducktape 框架的适配 Service、系统测试 case 等；
ꔷ spec：trogdor 的一些简单负载 example；
ꔷ unit：一些极简的测试，目前基本没有使用；
注：trogdor 是 Kafka 工程中定义的一个测试框架，系统测试中会利用该框架跑一些复杂的负载场景。
进一步地，kafkatest 的子目录包括：
ꔷ benchmarks：性能测试的 case；
ꔷ sanity_checks：较为基础的 case，主要验证简单生产消费、Kafka node 重启等基本场景；
ꔷ services：ducktape 的基本测试资源为 Service，本目录下是 Kafka 集群中各种测试资源对 Service 的实现，其中最重要的就是 KafkaService （对应的是 Kafka 集群）；
ꔷ tests：Kafka 工程中各个模块的系统测试，包括 Kafka Client、Kafka Connect、Kakfa Core、Kafka Streams 和 Kafka Tools 等；
2.3 运行系统测试
Talk is cheap. 本小结将以“sanity_checks/test_verifiable_producer.py”中的“test_multiple_kraft_security_protocols” 测试 case 为例，实际跑一下系统测试。
在 AutoMQ 工程的主目录下，输入以下命令触发上述系统测试：
TC_GENERAL_MIRROR_URL="mirrors.ustc.edu.cn" TC_PATHS="tests/kafkatest/sanity_checks/test_verifiable_producer.py::TestVerifiableProducer.test_multiple_kraft_security_protocols" bash tests/Docker/run_tests.sh
其中 TC_GENERAL_MIRROR_URL 的设置是为了在国内环境下加速 Docker image 的打包，TC_PATHS 指定系统测试 case。
如果你此前没有打包过系统测试的 Docker image，上述命令会自动触发打包逻辑。之后自动触发指定的系统测试，测试的输出类似于：
206 actionable tasks: 206 up-to-date
Docker exec ducker01 bash -c "cd /opt/Kafka-dev && ducktape --cluster-file /opt/Kafka-dev/tests/Docker/build/cluster.json ./tests/kafkatest/sanity_checks/test_verifiable_producer.py::TestVerifiableProducer.test_multiple_kraft_security_protocols "
/usr/local/lib/python3.9/dist-packages/paramiko/transport.py:236: CryptographyDeprecationWarning: Blowfish has been deprecated and will be removed in a future release
"class": algorithms.Blowfish,
[INFO:2024-04-25 08:13:03,399]: starting test run with session id 2024-04-25--002...
[INFO:2024-04-25 08:13:03,400]: running 4 tests...
[INFO:2024-04-25 08:13:03,400]: Triggering test 1 of 4...
[INFO:2024-04-25 08:13:03,418]: RunnerClient: Loading test {'directory': '/opt/Kafka-dev/tests/kafkatest/sanity_checks', 'file_name': 'test_verifiable_producer.py', 'cls_name': 'TestVerifiableProducer', 'method_name': 'test_multiple_kraft_security_protocols', 'injected_args': {'inter_broker_security_protocol': 'PLAINTEXT', 'metadata_quorum': 'ISOLATED_KRAFT'}}
[INFO:2024-04-25 08:13:03,422]: RunnerClient: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT: on run 1/1
[INFO:2024-04-25 08:13:03,425]: RunnerClient: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT: Setting up...
[INFO:2024-04-25 08:13:03,426]: RunnerClient: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT: Running...
[INFO:2024-04-25 08:13:37,574]: RunnerClient: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT: Tearing down...
[INFO:2024-04-25 08:13:59,432]: RunnerClient: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT: PASS
[INFO:2024-04-25 08:13:59,434]: RunnerClient: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT: Data: None
[INFO:2024-04-25 08:13:59,447]: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
...
================================================================================
SESSION REPORT (ALL TESTS)
ducktape version: 0.11.4
session_id: 2024-04-25--002
run time: 3 minutes 50.640 seconds
tests run: 4
passed: 4
flaky: 0
failed: 0
ignored: 0
================================================================================
test_id: kafkatest.sanity_checks.test_verifiable_producer.TestVerifiableProducer.test_multiple_kraft_security_protocols.inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT
status: PASS
run time: 56.010 seconds
--------------------------------------------------------------------------------
...
可以看到成功触发了 4 个系统测试，测试全部成功。
在工程的主目录下的 results 目录下，可以看到测试的详细产物：
. ├── TestVerifiableProducer #测试类 │ └── test_multiple_kraft_security_protocols # 测试类中的方法 │ ├── inter_broker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT # 测试方法的参数信息 │ │ └── 1 │ │ ├── KafkaService-0-281473792730304 # controller节点群 │ │ │ └── ducker02 │ │ │ ├── info # 日志信息 │ │ │ │ ├── controller.log │ │ │ │ ├── Kafka-authorizer.log │ │ │ │ ├── Kafka-request.log │ │ │ │ ├── log-cleaner.log │ │ │ │ ├── server.log │ │ │ │ └── state-change.log │ │ │ └── server-start-stdout-stderr.log │ │ ├── KafkaService-1-281473792730160 # broker节点群 │ │ │ └── ducker03 │ │ │ ├── info │ │ │ │ ├── controller.log │ │ │ │ ├── Kafka-authorizer.log │ │ │ │ ├── Kafka-request.log │ │ │ │ ├── log-cleaner.log │ │ │ │ ├── server.log │ │ │ │ └── state-change.log │ │ │ └── server-start-stdout-stderr.log │ │ ├── MiniKdc-0-281473793298144 │ │ │ └── ducker05 │ │ │ └── minikdc.log │ │ ├── VerifiableProducer-0-281473792730400 # 消费者日志 │ │ │ └── ducker04 │ │ │ └── verifiable_producer.log │ │ ├── report.json │ │ ├── report.txt │ │ ├── test_log.debug │ │ └── test_log.info # 测试过程的 std 信息 ├── report.css ├── report.html # 测试的整体结果报告 ├── report.json ├── report.txt ├── report.xml ├── session_log.debug └── session_log.info
对于成功的测试，只会保留 info 级别日志；对于失败的测试，还会保留 debug 级别的日志。
使用以下命令销毁测试容器：
./tests/Docker/ducker-ak down
2.4 细节信息
默认情况下，系统测试会启动 14 个 ducker 容器 ducker01 ~ ducker14(前文中打包的测试镜像的实例) + 1 个 S3 服务容器。所有 ducker 容器只负责一个职能，即作为 Kafka 节点、 producer、consumer 等。其中 ducker01 为管控容器，负责拉起关闭 Kafka 节点、拷贝日志之类的操作。S3 服务容器使用了 localstack[3] 在本地模拟 S3 存储。
每个 ducker 容器会将宿主机上的工程主目录映射到容器的 /opt/Kafka-dev 下，默认测试的都是开发版本的 Kafka 代码。在 ducker 容器的 /opt 目录下，还可以看到其他版本的 Kafka 产物：
ducker@ducker01:/opt$ ls kafka-0.10.0.1 kafka-0.10.2.2 kafka-0.8.2.2 kafka-1.0.2 kafka-2.0.1 kafka-2.2.2 kafka-2.4.1 kafka-2.6.3 kafka-2.8.2 kafka-3.1.2 kafka-3.3.2 kafka-3.5.2 kafka-3.7.0 kibosh kafka-0.10.1.1 kafka-0.11.0.3 kafka-0.9.0.1 kafka-1.1.1 kafka-2.1.1 kafka-2.3.1 kafka-2.5.1 kafka-2.7.2 kafka-3.0.2 kafka-3.2.3 kafka-3.4.1 kafka-3.6.1 kafka-dev
在某些系统测试中，这些产物会用来验证 Kafka 各个版本的客户端和服务端能够互相兼容。
系统测试的过程大致可以分为以下步骤：
-
测试扫描：统计本次运行涉及的所有 case；
-
节点分配：从 14 个 ducker 容器中分配节点，这些节点将负责执行 Service 的实现（KafkaService、 VerifiableProducer、MiniKdc 等）；
-
Service 的启动：为了保证幂等性，一般是先做现场的清理（stop service、日志删除等），然后按照系统测试的配置启动 Service。
-
执行测试：所有 Service ready 以后，执行系统测试的详细逻辑；
-
收尾：一次测试结束后，记录测试的结果、拷贝测试过程中的日志，并清理现场（stop service、日志删除等）
需要注意的是，上述过程均由 ducker01 容器管控，管控的方式是 SSH 免密登录到目标容器，然后执行命令。集成测试的验证逻辑，例如生产消息的数目是否预期，也是基于 SSH 远程查询、解析来实现的。
另外，所有测试是串行执行的，会复用这 14 个 ducker 容器 + 1 个 S3 服务容器。
2.5 参数化测试
在软件测试中，参数化测试是一种常见的策略，它允许开发者使用不同的输入数据来执行相同的测试用例，从而实现更全面的覆盖。这种方法非常适合于需要测试多种输入组合的情况。Kafka的系统测试采用了这种优秀的测试参数注入设计。以其系统测试背后的一个例子为例，我们可以看到测试方法上有两种注解：
ꔷ @cluster：指定测试涉及的最大节点数；
ꔷ @matrix：测试参数注入；
@cluster(num_nodes=4)
@matrix(inter_broker_security_protocol=['PLAINTEXT', 'SSL'], metadata_quorum=[quorum.isolated_kraft])
@matrix(inter_broker_security_protocol=['SASL_SSL'], inter_broker_sasl_mechanism=['PLAIN', 'GSSAPI'],
metadata_quorum=[quorum.isolated_kraft])
def test_multiple_kraft_security_protocols(
self, inter_broker_security_protocol, inter_broker_sasl_mechanism='GSSAPI', metadata_quorum=quorum.isolated_kraft)
@matrix注解用于定义参数的可能值，它可以看作是一个参数的集合。例如，@matrix(inter_broker_security_protocol=[‘PLAINTEXT’, ‘SSL’], metadata_quorum=[quorum.isolated_kraft]) 这个注解表示方法参数 inter_broker_security_protocol 可以分别取值 ‘PLAINTEXT’ 和 ‘SSL’，而 metadata_quorum 取值为 quorum.isolated_kraft。这样，就形成了一个2 * 1的测试矩阵，即两种可能的测试场景。
如果存在多个 @matrix 注解，那么会生成一个包含所有可能组合的测试矩阵。例如，在这个例子中，test_multiple_kraft_security_protocols 方法使用一份代码构造了21 + 12*1 = 4种测试场景。
这种矩阵式的测试参数方法有一个显著的优点，那就是它可以大大减少重复的代码。通过使用参数化测试，开发者可以用一份代码来测试许多不同的场景，从而提高代码的可重用性，同时也提高了测试的效率和覆盖率。
03 AutoMQ系统测试现状
AutoMQ 完全适配并继承了 Apache Kafka 的系统测试。由于 AutoMQ 使用 S3 作为主要存储， 所以系统测试中引入了 S3 服务容器，并在测试之间清理 S3 的数据（以保证测试隔离）。
系统测试很好地监督了 AutoMQ 的代码改动，确保其对 Apache Kafka 保证100%的兼容性。例如，ISSUE-751[4] 中，QuotaTest 及时发现了此前 AutoMQ 对网络层模型的修改带来的衍生问题。
AutoMQ 目前只支持 KRaft 模式下启动 Kafka 集群，因此我们更关注 KRaft 模式下的系统测试。除了适配已有的系统测试 case，在 2024 年 1 月，我们还全量改造了所有 case 以支持 KRaft 模式的测试：由于 Kafka 早期是基于 zk 集群运行的，大量系统测试还是仅限 zk 模式下运行的，我们梳理并改造了相关测试，以支持 KRaft 模式下的验证。
截止到 2024 年 4 月，AutoMQ 共有 543 个系统测试案例（其余基本都是 zk 模式的测试，或者已经废弃的测试）。自 2023 年以来，系统测试每日由 github action 触发，并自动报告测试的结果：
注：上图是 Kafka Connect 模块中系统测试概览报告，并不是全量的报告。你也可以通过 [5] 和 [6] 查看 AutoMQ main 分支下的代码每天全量系统测试的结果。
利用系统测试，AutoMQ 可以确保 0.9.x 版本以来的 Kafka 客户端与 AutoMQ 的兼容性，以及 Kafka Connect 等衍生产品与 AutoMQ 的兼容性。此外还可以验证真实用户场景（事务消息、消费者动态平衡）、灾难场景（Kafka node 宕机、 进程卡死）下 AutoMQ 服务的健壮性。
04 总结
本文主要介绍了 Kafka 的系统测试框架，并简要介绍了系统测试的原理。AutoMQ 完全继承了 Apache Kafka 丰富的系统测试案例，模拟真实用户使用 Kafka 集群的行为，并验证行为的结果是否符合预期。500+ 系统测试案例在一定程度上保证了 AutoMQ 代码的健壮性，同时能够保证 AutoMQ 对 Apache Kafka 的 100% 兼容性。
引用
[1]ducktape: https://ducktape.readthedocs.io/en/latest/ [2] AutoMQ: https://github.com/AutoMQ/AutoMQ [3] localstack: https://github.com/localstack/localstack [4] ISSUE-751: https://github.com/AutoMQ/automq/issues/751 [5] main E2E tests：Nightly Main E2E tests · Workflow runs · AutoMQ/AutoMQ [6] extra E2E tests: Nightly Main E2E tests · Workflow runs · AutoMQ/AutoMQ
关于我们
我们是来自 Apache RocketMQ 和 Linux LVS 项目的核心团队，曾经见证并应对过消息队列基础设施在大型互联网公司和云计算公司的挑战。现在我们基于对象存储优先、存算分离、多云原生等技术理念，重新设计并实现了 Apache Kafka 和 Apache RocketMQ，带来高达 10 倍的成本优势和百倍的弹性效率提升。
🌟 GitHub 地址：https://github.com/AutoMQ/automq
💻 官网：https://www.automq.com
👀 B站：AutoMQ官方账号
🔍 视频号：AutoMQ
-
MQ-2烟雾传感器详解12-21
-
Kafka消息丢失资料：新手入门指南12-09
-
Kafka消息队列入门：轻松掌握Kafka消息队列12-07
-
Kafka消息队列入门：轻松掌握消息队列基础知识12-07
-
Kafka重复消费入门：轻松掌握Kafka消费的注意事项与实践12-07
-
Kafka重复消费入门教程12-07
-
RabbitMQ入门详解：新手必看的简单教程12-07
-
RabbitMQ入门：新手必读教程12-07
-
Kafka解耦学习入门教程12-06
-
Kafka入门教程：快速上手指南12-06
-
Kafka解耦入门教程：实现系统间高效通信12-06
-
Kafka消息队列入门教程：从零开始轻松掌握12-06
-
RabbitMQ入门指南：轻松搭建与使用教程12-06
-
知乎启用AutoMQ替换Kafka，开辟成本优化与运维提效新纪元12-02
-
海外开发者为 AutoMQ 写的精品介绍，太干了！11-27
