在 Spring Boot 中使用 Kafka 测试容器时，如何动态覆盖 Bootstrap 服务器的配置？

【问题标题】：How Do I Override the Bootstrap Servers' Configuration Dynamically when Using Kafka Testcontainers in Spring Boot?在 Spring Boot 中使用 Kafka 测试容器时，如何动态覆盖 Bootstrap 服务器的配置？
【发布时间】：2023-02-02 02:46:30
【问题描述】：

我正在使用 Kafka Testcontainers 进行集成测试。容器启动，并读取在 application.yml 中设置的引导服务器配置。

由于 Testcontainers 端口是动态创建的，我使用 kafkaContainer.getBootstrapServers() 来获取由测试容器创建的套接字。

根据Spring Kafka documentation 和几篇在线文章，DynamicPropertyRegistry 类可用于覆盖现有配置。

这是我用来覆盖引导服务器配置的代码。

@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
}

但是，上面的代码似乎没有效果，因为测试的输出显示了几行

Connection to node -1 (localhost/127.0.0.1:29092) could not be established. Broker may not be available.

然后最终失败并显示此错误消息：

org.springframework.kafka.KafkaException: Send failed

at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:731)
at org.springframework.kafka.core.KafkaTemplate.observeSend(KafkaTemplate.java:691)
at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:514)
at com.zeal.accelerator.kafka.service.KafkaService.sendMessage(KafkaService.java:38)
at com.zeal.accelerator.kafka.producer.Producer.sendAvroMessage(Producer.java:24)
at com.zeal.integration.accelerator.kafka.ProducerIT.verify_that_expected_event_is_successfully_sent(ProducerIT.java:96)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:727)
at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:156)
at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:147)
at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:86)
at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(InterceptingExecutableInvoker.java:103)
at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.lambda$invoke$0(InterceptingExecutableInvoker.java:93)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:92)
at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:86)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:217)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:213)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:138)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:68)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:151)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:35)
at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:54)
at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:147)
at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:127)
at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:90)
at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:55)
at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:102)
at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:54)
at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:114)
at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:86)
at org.junit.platform.launcher.core.DefaultLauncherSession$DelegatingLauncher.execute(DefaultLauncherSession.java:86)
at org.junit.platform.launcher.core.SessionPerRequestLauncher.execute(SessionPerRequestLauncher.java:53)
at com.intellij.junit5.JUnit5IdeaTestRunner.startRunnerWithArgs(JUnit5IdeaTestRunner.java:57)
at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: org.apache.kafka.common.errors.TimeoutException: Topic test-topic not present in metadata after 60000 ms.

这是测试类。

@Slf4j
@Testcontainers
@ActiveProfiles({"test"})
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = KafkaApplication.class)
class ProducerIT {

private final static String TEST_TOPIC = "test-topic";

private TestAvroMessage expectedPayload;

private final CountDownLatch latch = new CountDownLatch(1);

@Autowired
private KafkaService kafkaService;

@Autowired
private Producer kafkaProducer;

@Autowired
private KafkaProperties kafkaProperties;

@Container
private static final KafkaContainer kafkaContainer =
        new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"))
                .withEmbeddedZookeeper();

@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
}

@Test
void verify_that_expected_event_is_successfully_sent() throws ExecutionException, InterruptedException {

    ReflectionTestUtils.setField(kafkaService, "producerTopic", TEST_TOPIC);
    kafkaProperties.setBootstrapServers(List.of(kafkaContainer.getBootstrapServers()));



    TestAvroMessage actual = TestAvroMessage.newBuilder()
            .setId("aDemoId")
            .setName("test")
            .setProcessed(true)
            .setCost(125000.00)
            .setLevel(5)
            .build();

    kafkaProducer.sendAvroMessage(actual);

    latch.await(10, TimeUnit.SECONDS);

    assertThat(actual).isEqualTo(expectedPayload);
}

@KafkaListener(topics = TEST_TOPIC,
        groupId = "${test.kafka.group-id}",
        containerFactory = "eventListenerContainerFactory")
void messageListener(TestAvroMessage event,
                     @Header(value = KafkaHeaders.RECEIVED_KEY, required = false) String messageKey,
                     @Header(KafkaHeaders.RECEIVED_PARTITION) String partitionId,
                     @Header(KafkaHeaders.OFFSET) long offset,
                     @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                     @Headers Map<String, Object> headers) throws InterruptedException {

    latch.countDown();

    expectedPayload = event;
}
}

我正在寻求帮助解决这个问题。架构要求包括使用 Testcontainer 并排除使用 EmbeddedKafka。谢谢！

【问题讨论】：

    标签： java spring-boot apache-kafka integration-testing testcontainers-junit5


    【解决方案1】：

    我们（Kent Hervey 和我）找到了如下答案：

    • @DynamicPropertySource 肯定被执行了。
    • 属性 spring.kafka.bootstrap-servers 不正确。我们无法在 Spring Kafka documentation 中找到它。我们使用调试器逐步检查并确定哪个属性键可用。
    • 更改属性键后，测试能够成功连接到在测试容器中运行的 Kafka 代理。
    • 我们现在正在做的是模式注册表测试容器（它不存在，所以我们正在扩展 GenericContainer）。
    • 我们仍然在 KafkaTemplate 上调用相同的方法，因为它实际上是在服务类中调用的。 spring.kafka.properties.bootstrap.servers

    以下代码反映了我们为动态设置 Kafka Bootstrap 服务器所做的更改。

    @DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
    registry.add("spring.kafka.properties.bootstrap.servers", kafkaContainer::getBootstrapServers);
}

    【讨论】：

      猜你喜欢
      • 2018-05-06
      • 2019-12-07
      • 1970-01-01
      • 2014-12-20
      • 2020-08-08
      • 1970-01-01
      • 2021-03-21
      • 1970-01-01
      • 2019-12-19
      相关资源
      最近更新 更多
      热门标签
      Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode