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队列的任何方式，其行为都应与本地集群相同-即为主题队列中的键插入提供重复数据删除.

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特定功能或具有kakfa特定功能的实现，那么唯一的方法就是使用kakfa！

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

kafka是否对其重复数据删除逻辑有任何测试?如果是这样，则以下各项的组合可能足以减轻您的组织认为的失败风险:

Does kafka have any tests around its deduplication logic? If so, the combination of the following maybe enough to mitigate your organizations 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. Once the initial legwork of creating a kafka docker image (one is probably already available from the community) it becomes trivial to setup integration test pipelines. A pipeline could look like:

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

我认为重要的是确保最小化kafka集成测试，以仅包括绝对依赖于kafka特定功能的测试.即使使用docker compose，它们也可能比单元测试慢几个数量级，大约是1毫秒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屋！