嵌入式卡夫卡测试随机失败

Question

我使用EmbededKafka实现了一系列集成测试，以测试一个使用spring框架运行的Kafka应用程序。

流应用程序读取来自Kafka主题的消息，将其存储到内部状态存储中，进行一些转换，并将其发送到另一个微服务以请求主题。当响应返回到响应主题时，它从状态存储中检索原始消息，并根据某些业务逻辑将其转发到我们的下游系统之一，每个下游系统都有自己的主题。

集成测试只是执行业务条件的各种排列。

最初，测试被分成几个类。在运行构建时，来自一个类的测试与另一个类中的测试发生冲突，但有一些冲突异常。我没有花太多时间在这个问题上，只是把所有的测试都移到了同一个类中。这解决了我的问题，所有测试都通过从gradle构建或从intelij。

下面是一个测试：

package au.nab.tlm.streams.integration;

import au.nab.tlm.streams.serde.EntitlementsCheckSerDes;
import au.nab.tlm.streams.test.support.MockEntitlementsCheckSerDes;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;

@SpringBootTest
@ContextConfiguration(classes = {MyTopologiesIntegrationTest.TestKafkaConfig.class})
@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
@EmbeddedKafka(
        ports = 9092,
        partitions = 1,
        topics = {
                "topic-1.v1",
                "topic-2.v1",
                "topic-3.v1",
                "topic-4.v1",
                "topic-5.v1",
                "topic-6.v1",
        },
        brokerProperties = {"transaction.state.log.replication.factor=1", "transaction.state.log.min.isr=1", "log.dir=/tmp/embedded-kafka"}
)
public class MyTopologiesIntegrationTest {
    @Autowired
    EmbeddedKafkaBroker kafkaBroker;

    @Autowired
    EntitlementsCheckSerDes appSerDes;

    @Test
    public void test_1() {
    }

    @Test
    public void test_2() {
    }

    @Test
    public void test_3() {
    }

    @Test
    public void test_4() {
    }

    @Test
    public void test_5() {
    }

    @TestConfiguration
    public static class TestKafkaConfig {
        @Bean
        EntitlementsCheckSerDes appSerDes() {
            return new MockEntitlementsCheckSerDes();
        }
    }
}

对结果感到满意的是，我推动了我的更改，只是为了注意到我们的CI服务器上的构建失败了。再次运行它，这一次它再次失败，与第一次不同的失败。我让一位同事看了看，他有着与CI服务器相似的失败体验。我在我的机器上运行了至少20次，它总是经过。在我的同事身上一个接一个地进行测试也总是能通过。

我们得到的最常见的例外是，主题xyz已经存在，但偶尔也会有其他一些例外，表明集群不可能喜欢或类似。所有这些异常都向我们表明，aa上一次测试中使用的嵌入式Kafka在下一次测试开始之前并没有完全关闭，尽管使用了DirtiesContext annotation.The，但是第一次运行的测试总是通过的。

我们都花了整整一天的时间脱发，这是不可能的。不管谷歌带我们去哪里，我们都试过了，一点运气都没有。最后，我们在测试类中留下了唯一的一个测试场景(交互次数最多的测试场景)，并禁用了其余的测试场景。

显然，这不是一个永久的解决办法，我真的想了解我们做错了什么，以及如何解决这个问题。

预先感谢您的投入。

Answer 1

不要使用固定端口ports = 9092, --默认情况下，嵌入式卡夫卡将监听操作系统选择的随机端口。

您应该在测试用例中使用它。

您可以通过调用this.kafkaBroker.getBrokersAsString()获得代理地址。