问题标签 [spring-kafka-test]

我已经配置了一个带有 transactionIdPrefix 的 Kafka ProducerFactory，以便使用@Transactional（请参阅Spring 文档关于仅生产者事务）启用事务同步。

我在集成测试中运行 EmbeddedKafka，以查看它的行为方式。

日志显示以下内容：

这是在 DefaultKafkaProducerFactory 执行时抛出的newProducer.initTransactions()。

我的配置如下：

集成测试

生产者配置

Spring-Kafka 版本：2.2.7.RELEASE

我不知道如何继续前进，我认为我遵循了文档中的每一步，并且在事务初始化期间 Kafka 客户端和代理之间的通信应该没问题。谁能帮我解决这个问题？

我正在使用嵌入式 Kafka 对 Spring Boot 应用程序进行单元测试。我正在使用代码 - https://github.com/spring-projects/spring-kafka/blob/29a3bd2021c49b700d4f3835c7ced642322c2faf/spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.java# L76作为参考。

这是我的代码：

我在 EmbeddedKafkaCondition.getBroker() 行收到以下异常

我想我错过了一个设置，我的嵌入式Kafka 设置不正确，因此为 NULL。但我一直无法弄清楚原因。如果有人可以帮助我了解我所缺少的内容，那将会有很大帮助。

我正在为一个简单的 Spring Boot 应用程序编写一个 Kafka 集成测试。应用程序简单地发布到 Kafka 主题。

我正在使用嵌入式 Kafka 实例进行测试。当通过 Intellij 运行时，测试工作得很好，但当我通过 gradle 运行它时失败。看起来latch倒计时永远不会达到 0，并且测试最终会超时。

生产者配置：

制片人：

消费者：

测试：

应用程序.yml：

知道是什么问题吗？

为了测试我们的 Spring Boot 应用程序在 kafka 集群尚未启动时如何处理它，我想在应用程序启动后的某个时间在 junit 测试中启动一个嵌入式 kafka 集群。我怎么能接近这个？据我了解，spring-kafka-test@EmbeddedKafka在创建 SpringBootTest 的应用程序上下文之前启动集群。有没有办法配置那个时间？

当尝试使用@EmbeddedKafkafrom时，我在运行测试时spring-kafka-test最终得到ClassNotFoundException: scala.collection.convert.AsJavaExtensions(full stack trace here )。

我尝试了不同的 Kafka 版本（org.apache.kafka:kafka_2.11:2.4.1而不是org.apache.kafka:kafka_2.12:2.8.0）以及不同的版本org.scala-lang:scala-library（2.13.6，，，，2.12.10）。2.11.122.10.6

为了解决这个问题，我需要做什么？

以下是重现该问题的最小示例：

src/main/kotlin/com/acme/Application.kt

src/test/kotlin/com/acme/integration/TestFullStack.kt

build.gradle.kts

settings.gradle.kts

Dockerfile

我在 Spring Boot 中有一个这样的 Kafka 发布者。

这是我使用Mockk库的单元测试Kottest，Jnit5

但是我收到了这个错误

理想情况下，我想断言回调..有什么方法可以实现这一点吗？

我有一个具有以下配置的 kafka 属性的测试用例：

这是我在运行上面的测试用例时遇到的构建错误：

这是具有 spring kafka 和 kafka-client 依赖项的 POM 文件，它可能会导致错误：

解决方法：请参考这个链接： https ://docs.spring.io/spring-kafka/docs/current/reference/html/#update-deps

我已将 spring boot 版本从 2.5.3 升级到 2.5.5，它已经升级了这个库。

通过这次升级，我开始收到一个错误：java.lang.NoSuchMethodError:kafka.server.KafkaServer.(Lkafka/server/KafkaConfig; Lorg/apache/common/kafka/utils/Time;Lscala/Option;/Lscala/collection/序列；）V

我已经尝试了很多根据spring docs的兼容性矩阵的要求更改版本https://spring.io/projects/spring-kafka

知道可能是什么问题吗？

pom.xml (spring-boot 2.5.5)

我正在为连接到 Kafka 以使用和发布数据的应用程序编写集成测试，为此我正在使用 EmbeddedKafka。部分逻辑是使用特定偏移量的消息。我想模拟这个，因此我的目标是：

1. 向 EmbeddedKafka 发送一些消息，但具有特定的偏移量
2. 以相同的偏移量使用它们

这现在不起作用，即我正在发送带有 的消息KafkaHeaders.OFFSET，但它被忽略了，我之后使用的消息具有不同的偏移量。事实上，偏移量只是从 0 开始，然后递增。

KafkaTemplate<String, String>以标准方式初始化。另一方面，我以标准方式消费：

简而言之，接收到的消息的偏移量与onMessage消息构建期间设置的偏移量不同。

有什么办法可以做到这一点？

当我们调用这个。这是使用 Postgres 数据库。

在这个 Workflow 类中，这是一个下面的列

现在的情况是，当我添加 kafka-test 依赖项时，这个 Object 类型的输入值被转换为 HashTrieMap，但是当我删除这个依赖项时，我得到了 HashMap。

Spring Boot 版本：2.2.6.RELEASE

我正在尝试添加 Embedded Postgres 嵌入式 kafka 和嵌入式 redis。