spring-boot - 如何使用 EmbeddedKafkaRule/EmbeddedKafka 设置 Spring Kafka 测试以修复 TopicExistsException 间歇性错误?
问题描述
我在测试我的 Kafka 消费者和生产者时遇到了问题。集成测试间歇性地失败,出现TopicExistsException
.
这就是我当前的测试类UserEventListenerTest
的样子——对于其中一位消费者来说:
@SpringBootTest(properties = ["application.kafka.user-event-topic=user-event-topic-UserEventListenerTest",
"application.kafka.bootstrap=localhost:2345"])
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class UserEventListenerTest {
private val logger: Logger = LoggerFactory.getLogger(javaClass)
@Value("\${application.kafka.user-event-topic}")
private lateinit var userEventTopic: String
@Autowired
private lateinit var kafkaConfigProperties: KafkaConfigProperties
private lateinit var embeddedKafka: EmbeddedKafkaRule
private lateinit var sender: KafkaSender<String, UserEvent>
private lateinit var receiver: KafkaReceiver<String, UserEvent>
@BeforeAll
fun setup() {
embeddedKafka = EmbeddedKafkaRule(1, false, userEventTopic)
embeddedKafka.kafkaPorts(kafkaConfigProperties.bootstrap.substringAfterLast(":").toInt())
embeddedKafka.before()
val producerProps: HashMap<String, Any> = hashMapOf(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaConfigProperties.bootstrap,
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to "org.apache.kafka.common.serialization.StringSerializer",
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to "com.project.userservice.config.MockAvroSerializer"
)
val senderOptions = SenderOptions.create<String, UserEvent>(producerProps)
sender = KafkaSender.create(senderOptions)
val consumerProps: HashMap<String, Any> = hashMapOf(
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafkaConfigProperties.bootstrap,
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to "org.apache.kafka.common.serialization.StringDeserializer",
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to kafkaConfigProperties.deserializer,
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
"schema.registry.url" to kafkaConfigProperties.schemaRegistry,
ConsumerConfig.GROUP_ID_CONFIG to "test-consumer"
)
val receiverOptions = ReceiverOptions.create<String, UserEvent>(consumerProps)
.subscription(Collections.singleton("some-topic-after-UserEvent"))
receiver = KafkaReceiver.create(receiverOptions)
}
}
// Some tests
// Not shown as they are irrelevant
...
...
...
该类UserEventListener
使用来自 的消息user-event-topic-UserEventListenerTest
并将消息发布到some-topic-after-UserEvent
。
正如您从设置中看到的那样,我有一个测试生产者将向其发布消息user-event-topic-UserEventListenerTest
以便我可以测试是否UserEventListener
使用该消息和一个测试消费者将从该消息使用该消息some-topic-after-UserEvent
以便我可以查看是否将消息发布UserEventListener
到some-topic-after-UserEvent
处理记录。
KafkaConfigProperties
课程如下。
@Component
@ConfigurationProperties(prefix = "application.kafka")
data class KafkaConfigProperties(
var bootstrap: String = "",
var schemaRegistry: String = "",
var deserializer: String = "",
var userEventTopic: String = "",
)
application.yml
看起来像这样。
application:
kafka:
user-event-topic: "platform.user-events.v1"
bootstrap: "localhost:9092"
schema-registry: "http://localhost:8081"
deserializer: com.project.userservice.config.MockAvroDeserializer
错误日志
com.project.userservice.user.UserEventListenerTest > initializationError FAILED
kafka.common.KafkaException:
at org.springframework.kafka.test.EmbeddedKafkaBroker.createTopics(EmbeddedKafkaBroker.java:354)
at org.springframework.kafka.test.EmbeddedKafkaBroker.lambda$createKafkaTopics$4(EmbeddedKafkaBroker.java:341)
at org.springframework.kafka.test.EmbeddedKafkaBroker.doWithAdmin(EmbeddedKafkaBroker.java:368)
at org.springframework.kafka.test.EmbeddedKafkaBroker.createKafkaTopics(EmbeddedKafkaBroker.java:340)
at org.springframework.kafka.test.EmbeddedKafkaBroker.afterPropertiesSet(EmbeddedKafkaBroker.java:284)
at org.springframework.kafka.test.rule.EmbeddedKafkaRule.before(EmbeddedKafkaRule.java:114)
at com.project.userservice.user.UserEventListenerTest.setup(UserEventListenerTest.kt:62)
Caused by:
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'user-event-topic-UserEventListenerTest' already exists.
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:104)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:272)
at org.springframework.kafka.test.EmbeddedKafkaBroker.createTopics(EmbeddedKafkaBroker.java:351)
... 6 more
Caused by:
org.apache.kafka.common.errors.TopicExistsException: Topic 'user-event-topic-UserEventListenerTest' already exists.
我试过的:
- 通过指定引导配置,在每个测试中使用不同的引导服务器地址,例如
@SpringBootTest(properties = ["application.kafka.bootstrap=localhost:2345"])
- 通过覆盖主题配置,在每个测试中使用不同的主题名称,
@SpringBootTest
就像上一个要点中的引导服务器覆盖一样 - 添加
@DirtiesContext
到每个测试类
软件包版本
- 科特林 1.3.61
- Spring Boot - 2.2.3.RELEASE
- io.projectreactor.kafka:reactor-kafka:1.2.2.RELEASE
- org.springframework.kafka:spring-kafka-test:2.3.4.RELEASE (仅测试实现)
问题
EmbeddedKafkaRule
我有多个使用并设置或多或少相同的测试类。对于它们中的每一个,我指定了不同的 kafka 引导服务器地址和主题名称,但我仍然间歇性地看到 TopicAlreadyExists 异常。
我可以做些什么来使我的测试类保持一致?
解决方案
我指定了不同的 kafka 引导服务器地址和主题名称,但我仍然间歇性地看到 TopicAlreadyExists 异常
这是没有意义的; 如果他们每次都有一个新的端口,尤其是新的主题名称,那么主题就不可能已经存在。
一些建议:
- 由于您使用的是 JUnit5,请不要使用 JUnit4
EmbeddedKafkaRule
,EmbeddedKafkaBroker
而是使用;或者简单地添加@EmbeddedKafka
,代理将作为 bean 添加到 Spring 应用程序上下文及其由 Spring 管理的生命周期(用于@DirtiesContext
销毁);对于非 Spring 测试,代理将由 JUnit5 创建(和销毁),EmbeddedKafkaCondition
并可通过EmbeddedKafkaCondition.getBroker()
. - 不要使用显式端口;让代理使用其默认随机端口并
embeddedKafka.getBrokersAsString()
用于引导服务器属性。 - 如果您必须自己管理经纪人(在 中
@BeforeAll
),destroy()
他们在@AfterAll
.
推荐阅读
- javascript - React-Router-Dom 滚动到组件在移动设备上无法正常工作
- ansible - 基于正则表达式的合并字典
- c++ - 指向对象的指针的 C++ 重载运算符
- reactjs - 无限滚动两次更新数据
- node.js - 我的 req.session 数据不是全局的,就像我尝试通过 app.get 从服务器发送它时显示 undefiend - express session
- python - Windows CMD:“此 Python 解释器处于 conda 环境中,但尚未激活该环境”
- java - 用eclipse对servlet进行Junit测试
- javascript - 在嵌套对象中搜索键/值 (JavaScript)
- javascript - React 无限滚动组件
- flutter - 无法在 GestureDetector 上使用 longPress 可靠地触发动画