问题标签 [spring-kafka-test]

这是刚刚启动的带有简单主类的 spring boot 项目，它可以在没有 spring-kafka 依赖的情况下正常工作，但是在添加spring-kafka并spring-kafka-test爆炸后异常，github here

gradle.build

主班

错误

当使用 Integer 作为键时，这不是问题，kafka 应该能够将字符串作为键来处理。

它抛出以下异常：

Org.apache.kafka.common.errors.SerializationException：无法将类 java.lang.String 的键转换为 key.serializer 中指定的类 org.apache.kafka.common.serialization.IntegerSerializer

原因：java.lang.ClassCastException：java.lang.String 无法在 org.apache.kafka.common.serialization.IntegerSerializer.serialize(IntegerSerializer.java:21) 的 org.apache.kafka 中转换为 java.lang.Integer .common.serialization.ExtendedSerializer$Wrapper.serialize(ExtendedSerializer.java:65) at org.apache.kafka.common.serialization.ExtendedSerializer$Wrapper.serialize(ExtendedSerializer.java:55) at org.apache.kafka.clients.producer .KafkaProducer.doSend(KafkaProducer.java:799) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:784) at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java :285) 在 org.springframework.kafka 的 org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:357)。core.KafkaTemplate.send(KafkaTemplate.java:206)

我目前正在使用spring boot 2和junit 5。我想使用spring-kafka-test进行测试，但是我使用的spring-kafka版本是2.1.8.RELEASE。有了这个，我相信它不可能使用EmbeddedKafkaBroker，因为这是 2.2.x 中的东西。

我使用的 kafka-client 是 1.1.0。我查看了https://spring.io/projects/spring-kafka中提到的兼容性矩阵，它似乎不适用于客户端 1.1.0，只是想知道是否可以仅将 spring-kafka-test 更新为2.2.3.RELEASEEmbeddedKafkaBroker与 Junit 5 一起使用，同时仍使用旧的 kafka-client 1.1.0。

有没有人尝试过这样的事情？

我正在尝试对使用 kafka、kafka-streams 和 cassandra 的应用程序进行集成测试。但是当我尝试设置测试类时，我有 2 个错误： 错误 [main] BrokerMetadataCheckpoint: 无法读取 dir 下的 meta.properties 文件 错误 [main] KafkaServer: 无法读取日志目录下的 meta.properties

我正在使用 spring-boot-starter 2.1.2、spring-boot-starter-test 2.1.2、spring-kafka 2.2.0、spring-kafka-test 2.2.0、apache.kafka-streams 2.1.0

试图更改 logs.dir 和 logs.dirs 参数。使用@EnableKafka @EnableKafkaStreams

我希望使用嵌入式 kafka 运行上下文，但现在我收到一个错误，即 meta.properties 不存在

我正在为我的 Spring Boot 应用程序编写 Junit 测试用例（使用 @EmbeddedKafka），该应用程序广泛使用 Spring-kafka 与其他服务和其他操作进行通信。

一个典型的案例是从 kafka 中删除数据（我们在 kafka 中推送空消息）。

目前在 delete() 方法中，我们首先检查 kafka 中是否存在任何请求删除的消息。然后我们在 Kafka 中为该消息键推送null

为上述方法逻辑编写 Junit 的步骤。

这里的问题是 Kafka 总是抛出 message not found 异常。在 service.delete() 方法中。

在控制台中检查日志时。我发现我的生产者配置为 kafka 使用不同的端口，而消费者配置使用不同的端口。

我不确定我是否遗漏了一些细节，或者这种行为的原因是什么。任何帮助将不胜感激。

I have spring boot application that had a consumer consumes from topic in one cluster and produces to another topic in different cluster.

Now I'm trying to write integration test case using spring embedded Kafka but having an issue KafkaTemplate could not be registered. A bean with that name has already been defined in class path resource

Consumer Class

Producer Class

In my Test cases I want to override KafkaTemplate so that when i call kafkaConsumerService.professor method in Test it should produce the data into embedded Kafka and i should validate it.

Test config

Test class

Error

And also can some one help me how to validate messages sent into embedded Kafka server?

Note I'm having some deprecated warnings

The type KafkaEmbedded is deprecated

The method getPartitionsPerTopic() from the type KafkaEmbedded is deprecated

The method producerProps(KafkaEmbedded) from the type KafkaTestUtils is deprecated

我正在尝试使用spring-boot和 spring 嵌入式 Kafka 进行集成测试。我能够向 Spring 嵌入式 Kafka 服务器生成消息，但服务类中的侦听器不是测试侦听器，而是尝试使用记录和

KafkaProducerConfigTest包含所有 bean 的配置类

KafkaProducerServiceTest服务测试类，但是这里的监听器没有消耗任何数据，我知道这个测试会失败，因为这不是正确的方法

错误

根据@Gray Russel先生的回答进行了更新，但我仍然有问题

我在 Spring Boot 应用程序中使用 spring-kaka-2.2.0 编写集成测试，我几乎成功了，我的测试用例仍然返回 true，但之后我仍然看到多个错误。

测试配置

测试班

测试用例testReceive()通过但仍有多个错误消息

堆栈跟踪错误 1

堆栈跟踪错误 2

堆栈跟踪错误 3

我与kafka相关的测试依赖如下：

我在 junit 入口点类中添加规则：

但 UT 启动失败，但有以下异常：

在我使用 org.apache.kafka:kafka_2.11:0.10.1.1 和 spring-kafka-test:1.1.3.RELEASE 之前，它们运行良好。我不知道这是我在春季 kafka 测试中的新依赖问题或错误。

我正在使用带有@EmbeddedKafka注释的 Spring Kafka 来启动 Embedded kafka 实例。

依赖：

我已经将我的应用程序配置为连接到它：

我正在成功发送消息并使用消息。

我的问题是消息保留在某处，可能会出现在进行中的测试运行中。日志正在打印这些：

高代数表示 Kafka 在测试重新运行之间保留了我不想要的信息。

如何彻底清理嵌入式 Kafka 并重新开始？

编辑：

我遇到的问题是我错误地使用了 Spring Profiles。我的嵌入式 Kafka 配置有一个自定义注释注释。问题是我@ActiveProfiles("kafka")在复合注释中设置，也在@ActiveProfiles("dev")实际测试类中设置。我已将复合注释更新为：