问题标签 [spring-kafka-test]

我有以下代码将数据发送到 Kafka。我能够成功发送测试 OnSucess 回调，但我无法测试 onFailure() 方法

我收到以下错误

java.lang.ClassCastException: org.mockito.codegen.Throwable$MockitoMock$2137573915 不能转换为 org.springframework.kafka.core.KafkaProducerException

我创建了以下测试类来使用 AvroSerializer 生成事件。

但是当我使用发送事件时， kafkaTemplate().send(appEventsTopic, applicationEvent);我得到了以下异常。

当我使用 MockSchemaRegistryClient 为什么它试图查找架构？

我们有一个集成测试，我们使用 EmbeddedKafka 并为一个主题生成一条消息，我们的应用程序处理该消息，然后将结果发送到我们使用和断言输出的第二个主题。在 CI 中，这可能有 2/3 的时间有效，但我们会遇到KafkaTestUtils.getSingleRecord抛出异常的情况java.lang.IllegalStateException: No records found for topic（参见下面的 [1]）。

为了尝试解决这个问题，我ContainerTestUtils.waitForAssignment为注册表中的每个侦听器容器添加了（参见下面的 [2]）。在 CI 中成功运行几次后，我看到了一个新异常：java.lang.IllegalStateException: Expected 1 but got 0 partitions. 这让我想知道这是否真的是原始异常未找到记录的根本原因。

有什么想法可以帮助解决这里的随机故障吗？我将不胜感激有关如何进行故障排除的任何建议。

spring-kafka 和 spring-kafka-test v2.6.4。

编辑：添加newConsumer以供参考。

我们的设置示例：

我们正在使用@EmbeddedKafka，并且application-test.yml我们有

我可以看到那confluentBootstrapAddress是null。

但是，如果在调试模式下，我会检查这个

然后我可以看到该值存在，并且嵌入式 Kafka 确实已启动。

可能是什么问题？application.yml评估时刻与嵌入式代理启动时的生命周期/时刻不匹配？

我一直在尝试使用 EmbeddedKafka 创建集成测试，但是在尝试运行它时遇到了缺少依赖项的问题，这是错误：

由于缺少依赖项 org/I0Itec/zkclient/serialize/ZkSerializer，无法加载类 org.springframework.kafka.test.EmbeddedKafkaBroker

我看到一些东西说这与我的依赖关系有关，所以这是我的依赖关系：

所以，我的问题是，我错过了什么吗？

编辑

按照指示更改版本后，我得到了一个不同的错误：

创建名为“embeddedKafka”的 bean 时出错：调用 init 方法失败；嵌套异常是 java.lang.NoClassDefFoundError: scala/math/Ordering$$anon$7

我已经添加了 scala 依赖项，但仍然有同样的问题：

有没有人尝试为您的 kafkaProducer 服务编写一个失败的 Junit 测试用例，用于不同的用例，如下所示。

• 与 KAFKA 集群通信时失败。想知道如果 EmbededKafkaBroker 负责连接到卡夫卡，如何做到这一点。

• 主题授权失败

提前致谢。

我试图让两个消费者订阅一个EmbeddedKafkaBroker。第一个成功了，第二个失败了。@EmbeddedKafka和经纪人都@ClassRule失败了。

我期待两个消费者可以订阅一个EmbeddedKafkaBroker。春季卡夫卡测试有可能吗？

我在这里复制了这个：https ://github.com/yraydhitya/spring-kafka-test-multiple-consumers

我已按照以下文档https://www.confluent.io/blog/spring-kafka-can-your-kafka-consumers-handle-a-poison-pill/处理反序列化异常。它工作正常，消息被记录并继续前进，但每次我重新启动服务器时，都会再次记录错误消息。有没有一种方法可以在记录错误消息后跳过/确认错误消息，以便在重新启动服务器时不会再次被拾取

另外，注意：= 从上面，我在日志中看到它给出了一个警告，即给出了 spring.deserializer.value.delegate.class -> 但不是已知的配置。

也配置了这个

有人可以提供同样的建议吗

我正在使用带有 Kafka 和弹性的 Spring Boot 为 Groovy 创建一个演示应用程序。我在 Spock 测试中使用了 @EmbeddedKafka 注释，它们在本地工作得非常好；在 Windows 和 Ubuntu 上。它们通过运行或调试在 Intellij 中工作，没有问题。在我的外壳“./gradlew test”中尝试时也是如此。万事皆安。一旦我将它推送到 github.com，我的 github 操作就失败了。但它正在调用相同的命令。

动作定义：https ://github.com/besessener/GroovySpringBootKafkaElasticsearchDemo/blob/main/.github/workflows/test.yml

远程失败测试用例： https ://github.com/besessener/GroovySpringBootKafkaElasticsearchDemo/blob/main/src/test/groovy/me/spring/GroovyDemo/stream/KafkaSendAndReceiveTest.groovy

行动：https ://github.com/besessener/GroovySpringBootKafkaElasticsearchDemo/runs/3019862203?check_suite_focus=true

从操作输出来看，对我来说唯一看起来像错误的是：

我读了很多关于不使用静态端口进行 kafka 测试的内容。但这是我唯一的 kafka 测试，所以我不太明白应该如何发生冲突。此外，LEADER_NOT_AVAILABLE 可能是不存在主题的问题，或者消费者可能根本无法正确连接到代理。但我没有看到任何这些。

我仍然觉得它与 "localhost:9092" 作为 brokerProperties 更相关。使用 Github 操作时是否有这方面的问题？或者我还缺少什么？

我是 Kafka 的新手，我的模型类用户面临以下问题 [请求处理失败；嵌套异常是 org.apache.kafka.common.errors.SerializationException: Can't convert value of class model.User to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer] 根本原因是 java.lang .ClassCastException：类 model.User 无法转换为类 java.lang.String（model.User 在加载器“app”的未命名模块中；java.lang.String 在加载器“bootstrap”的模块 java.base 中）在 org .apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:28) ~[kafka-clients-2.7.1.jar:na] 在*

我怀疑这是由于在 KafkaConfiguration 中错误地导入了 StringSerializer 和 JSONSerializer 。下面是我的代码

1- Kafka配置

2- UserResource 类

3-用户类

谁能让我知道我哪里出错了？这是否与进口有关（如果是，正确的是什么）？

谢谢