unit-testing - Kafka Streams:流线程未能锁定状态目录
问题描述
我正在尝试测试我的 Kafka Streams 应用程序。我已经构建了一个简单的拓扑结构,我从输入主题中读取数据并将相同的数据存储在状态存储中。
我尝试使用 TopologyTestDriver 为这个拓扑编写单元测试。当我运行测试时,我遇到了以下错误。
org.apache.kafka.streams.errors.LockException: stream-thread [main] task [0_0] Failed to lock the state directory for task 0_0
at org.apache.kafka.streams.processor.internals.AbstractTask.registerStateStores(AbstractTask.java:197)
at org.apache.kafka.streams.processor.internals.StreamTask.initializeStateStores(StreamTask.java:275)
at org.apache.kafka.streams.TopologyTestDriver.<init>(TopologyTestDriver.java:403)
at org.apache.kafka.streams.TopologyTestDriver.<init>(TopologyTestDriver.java:257)
at org.apache.kafka.streams.TopologyTestDriver.<init>(TopologyTestDriver.java:228)
at streams.checkStreams.checkStreamsTest.setup(checkStreamsTest.java:99)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:24)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
....
我可以看到状态存储在 /tmp/kafka-streams 中本地创建,但不知何故,流线程无法锁定它。我搜索并发现这个错误可能是因为两个流线程试图访问它,一个有锁,所以另一个必须等待。但是我没有看到在我的代码中创建了两个流线程。我是这个 kafka 流及其测试的新手,我在这里遗漏了什么吗?
解决方案
TopologyTestDriver
不会创建任何后台线程,因此多线程(来自 KafkaStreams 本身)应该不是问题——但是,正如@BartoszWardziński 指出的那样,如果您的测试框架并行执行测试,并且您在不同的测试中使用相同的application.id
测试,它可能会导致锁定问题。
测试的建议是生成随机数application.id
以避免此问题。
推荐阅读
- r - 通过传递一行检查每一列的约束
- c# - 使用 DataContract Xml 序列化的问题 - KnownTypes
- r - 错误:包 'adagio' 的编译失败
- node.js - 容器不尊重进程的退出状态
- python - 请求标头给出 TypeError: 'EnvironHeaders' object is not callable
- sql - 具有来自另一个表的主键的临时表
- android - 不能将 ExoPlayer 视图布局包含到另一个布局中
- python - 如何解决不打印作业要我打印的正确内容的问题?
- java - 如何使用 Spring Boot 和 Maven 测试微服务
- ruby-on-rails - X-Frame-Option 配置在 Rails 中不起作用