Python:如何模拟单元测试的 kafka 主题? [英] Python: how to mock a kafka topic for unit tests?

问题描述

我们有一个消息调度程序，它会根据消息属性生成一个哈希键，然后将其放置在带有该键的 Kafka 主题队列中.

We have a message scheduler that generates a hash-key from the message attributes before placing it on a Kafka topic queue with the key.

这样做是为了重复数据删除.但是，我不确定如何在不实际设置本地集群并检查它是否按预期执行的情况下测试此重复数据删除.

This is done for de-duplication purposes. However, I am not sure how I could possibly test this deduplication without actually setting up a local cluster and checking that it is performing as expected.

在网上搜索用于模拟 Kafka 主题队列的工具没有帮助，我担心我可能会以错误的方式思考这个问题.

Searching online for tools for mocking a Kafka topic queue has not helped, and I am concerned that I am perhaps thinking about this the wrong way.

最终，无论用于模拟 Kafka 队列的任何内容，都应该以与本地集群相同的方式运行 - 即通过向主题队列插入 Key 来提供重复数据删除.

Ultimately, whatever is used to mock the Kafka queue, should behave the same way as a local cluster - i.e. provide de-deuplication with Key inserts to a topic queue.

有没有这样的工具?

推荐答案

如果您需要验证特定于 Kafka 的功能，或使用特定于 Kafka 的功能来实现，那么唯一的方法就是使用 Kafka！

If you need to verify a Kafka specific feature, or implementation with a Kafka-specific feature, then the only way to do it is by using Kafka!

>

Kafka 是否对其重复数据删除逻辑进行了任何测试?如果是这样，以下各项的组合可能足以减轻您的组织感知到的失败风险:

Does Kafka have any tests around its deduplication logic? If so, the combination of the following may be enough to mitigate your organization's perceived risks of failure:

• 散列逻辑的单元测试(确保相同的对象确实生成相同的散列)
• Kafka 主题重复数据删除测试(Kafka 项目内部)
• 飞行前烟雾测试验证您的应用与 Kafka 的集成

如果 Kafka 没有围绕其主题重复数据删除进行任何类型的测试，或者您担心破坏性更改，那么围绕 Kafka 特定功能进行自动检查非常重要.这可以通过集成测试来完成.我最近在基于 Docker 的集成测试管道方面取得了很大的成功.在创建 Kafka docker 镜像(一个可能已经从社区中获得)的初始工作之后，设置集成测试管道变得微不足道.管道可能如下所示:

If Kafka does NOT have any sort of tests around its topic deduplication, or you are concerned about breaking changes, then it is important to have automated checks around Kafka-specific functionality. This can be done through integration tests. I have had much success recently with Docker-based integration test pipelines. After the initial legwork of creating a Kafka docker image (one is probably already available from the community), it becomes trivial to set up integration test pipelines. A pipeline could look like:

• 执行基于应用程序的单元测试(哈希逻辑)
• 一旦这些通过，你的 CI 服务器就会启动 Kafka
• 执行集成测试，验证重复写入仅向主题发出一条消息.

我认为重要的是确保 Kafka 集成测试被最小化，只包括绝对依赖于 Kafka 特定功能的测试.即使使用 docker-compose，它们也可能比单元测试慢几个数量级，~1ms vs 1 秒?另一件需要考虑的事情是维护集成管道的开销可能值得相信 Kakfa 会提供它声称的主题重复数据删除.

I think the important thing is to make sure Kafka integration tests are minimized to ONLY include tests that absolutely rely on Kafka-specific functionality. Even using docker-compose, they may be orders of magnitude slower than unit tests, ~1ms vs 1 second? Another thing to consider is the overhead of maintaining an integration pipeline may be worth the risk of trusting that Kakfa will provide the topic deduplication that it claims to.

这篇关于Python:如何模拟单元测试的 kafka 主题?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！