问题标签 [spring-kafka-test]

我正在尝试测试@KafkaListener 正在使用的containerlistenerfactory 中定义的错误处理程序。

根据我要测试的侦听器抛出的不同异常，我有不同的重试计数是否正常工作。但是在它第一次抛出异常并转到错误处理程序之后，我得到了一个非法状态异常，所以它没有尝试在我正在编写的测试中重试。相同的代码在实际设置中有效。

这是我得到的例外。

异常后查找当前；嵌套异常是 org.springframework.kafka.listener.ListenerExecutionFailedException

我希望测试重试 10 次，然后在恢复中打印消息。但它不会重试，因为错误处理程序正在抛出非法状态异常。

有人可以建议吗？

*******更新的代码，我无法在 EmbeddedKafka 上使用 bootstrapServersProperty ="spring.kafka.bootstrap-servers，因为它一直在尝试访问 localhost:9092 并超时

这是我的测试文件配置

AppConfig***（从属性文件中读取引导地址）

我有一个 kafka 监听器，它抛出一个异常并转到 seektocurrenterrorhandler 进行处理。我正在尝试包装侦听器以添加闩锁并对其进行断言以进行测试。但它似乎不起作用。有人可以建议。

听众：

配置：

测试：

我有一个简单的 spring-boot KStream 拓扑，可以将字符串从小写转换为大写。我希望我的集成测试启动嵌入式 kafka，然后测试拓扑。我想知道是否可以使用 spring 编写这样的集成测试@EmbeddedKafka？

我已经看到了几个使用@EmbeddedKafka简单消费者的例子，@KafkaListener但没有看到任何使用 KStream 的例子。

我尝试测试以下拓扑以将传入的文本流从小写转换为大写。

这是拓扑：

测试拓扑的单元测试是：

我想编写一个集成测试，首先启动嵌入式 kafka 服务器，然后测试 UppercaseStream 拓扑。

我尝试了以下方法：

测试未通过断言。我不确定如何获取 kStreamPromoToUppercase bean。我不确定我是否尝试遵循正确的方法。

我最近尝试通过集成测试和嵌入式 Kafka 来覆盖我的功能，但在运行它们时遇到了某些问题。结果是由 引发的异常KafkaServer。

这是集成测试类：

这里是测试配置文件配置

这个异常正在抛出。

我认为我的项目依赖项没有什么奇怪的。

这是根pom.xml https://pastebin.com/gx4YSyBC

这是pom.xml包含嵌入式 Kafka 的有问题测试的部署模块 https://pastebin.com/1V9EZ530

请，如果您曾经遇到过同样的问题，请分享您的解决方案。这对我很有帮助。几天前我被它的问题困住了。

我正在编写测试用例来测试我的应用程序的 Kafka 消费-转换-生产循环。因此，我有效地从 sourceTopic-processing-sendMessage 消费到 Destination 主题。我正在编写这些测试用例来证明与 Kafka 的一次性消息传递，因为稍后我将添加其他失败案例。
这是我的配置：

我的问题是上述测试用例“testProducerABSuccess”不一致，断言有时会失败，有时会通过。我一直无法弄清楚为什么它们如此不一致。以上有什么问题。

编辑：16-12：

1. 用 consumerconfig.Auto_Offset_Reset_config-earliest 测试没有变化。第一个断言通过了大约 70% 的时间。第二个断言始终失败（0% 通过率）。

我正在尝试使用 Springboot 和 EmbeddedKafka 为单元测试设置一个类。我有两个主题，topicA 和 topicB，我将测试消息生成到 topicA 和 topicB。

所以这是我的课：

现在，如果我运行测试，produceIntoTopicB 测试会失败并出现以下错误：

而另一个测试失败并出现此错误：

我哪里错了？

我使用这种方法编写了这个流应用程序：一个StreamConfigs类：

和这个其他类

在类的kStream()方法中，MyStream有我想要测试的应用程序的流逻辑。

所以我写了这个类进行测试，使用嵌入式kafka

这是testStream()方法

这不起作用，因为我在这行代码上遇到错误

这是堆栈跟踪

如何获取StreamBuilder实例以便重用我kStream()的测试方法和嵌入式代理？如果我尝试使用 来创建StreamBuilder实例new，我的测试会停留在等待我的机器上的活动代理。

你能帮助我吗？

编辑：我认为这解决了问题：我以这种方式编辑类的setupKafka()方法MyStreamApplicationTest：

我有一个使用 Spring boot 和 Spring kafka 库创建的应用程序。现在我将 jsonschema2pojo-core 库添加到我的 pom 文件中。添加此 jar 文件后，我的测试用例失败并出现以下错误

原因：org.springframework.beans.factory.BeanCreationException：创建名为“embeddedKafka”的bean时出错：调用init方法失败；嵌套异常是 java.lang.NoClassDefFoundError: scala/runtime/java8/JFunction0$mcZ$sp 原因：java.lang.NoClassDefFoundError: scala/runtime/java8/JFunction0$mcZ$sp

我创建了一个单元测试来测试 Kafka 侦听器，如下所示。

消费者创建如下所示。

消费者工厂和容器工厂的创建如下所示。

来自 POM 的依赖项是，

但是，当我调用测试用例时，消息会发布到嵌入式 Kafka，但不会调用实际的侦听器。不确定测试设置有什么问题。

application.properties 如下所示，

我想为 spring kafka 生产者和消费者编写单元测试。我尝试了多种方法：

1. 嵌入式Kafka注解
2. 嵌入式Kafka规则
3. 嵌入式KafkaBroker等...

每次我遇到一个或另一个错误时，GitHub 上发布的所有示例似乎都没有运行。我也检查了 spring kafka 版本的兼容性。

有人可以分享一个最近编写并成功运行的示例代码库吗？