junit结束时强制停止spring boot

Question

实际上，我有一个Spring Boot应用程序，我使用Kafka嵌入而不使用SpringKafka，而是使用合流库。

我配置了自定义拓扑，当我启动jUnit时，流仍然是监听的，Spring引导服务器不会结束。

我尝试使用@DirtiesContext，但问题仍然存在。

对于启动jUnit我使用

@RunWith(SpringRunner.class)
@SpringBootTest

当消费者处于循环中时，我在控制台中看到以下消息：

无法建立到节点-1的连接。经纪人可能不可用。

提示？

谢啦

Answer 1

似乎出现错误是因为代理没有运行。但是KafkaStreams正在运行。

为了修复它，我认为您需要在测试时阻止KafkaStreams运行。

@SpringBootTest注释提供了一个属性类。您可以指定将注册为bean的类

您可以通过指定class属性来防止与KafkaStreams相关的bean被注册。

例如，可以如下测试。在这种情况下，不会发生上述问题，因为KafkaStreams相关bean没有注册。

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {ServiceImpl.class, CommonConfig.class})
public class SomeClassTest {
    @Autowired
    private ServiceImpl articleServiceImpl;

    // do test
}

使用@SpringBootTest时，最好只注册要测试的bean。

如果您不熟悉这种方法，请尝试嘲笑管理KafkaStreams对象的bean

例如，我创建以下bean来开始和结束KafkaStreams对象。

@Component
public class ManageableStream implements DisposableBean, InitializingBean {

    private final KafkaStreams kafkaStreams;

    public ManageableStream() {
        StreamsConfig config = buildStreamConfig();
        StreamsBuilder builder = new StreamsBuilder();

        KStream<String, String> inputStream = builder.stream("source.topic");
        inputStream.to("destination.topic");

        Topology topology = builder.build();
        kafkaStreams = new KafkaStreams(topology, config);
    }

    @Override
    public void destroy() throws Exception {
        kafkaStreams.close();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        kafkaStreams.start();
    }

    private StreamsConfig buildStreamConfig() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-stream-application");
        properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, AT_LEAST_ONCE);
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);

        return new StreamsConfig(properties);
    }
}

测试时，模拟管理KafkaStreams对象的bean。

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {

    @MockBean
    private ManageableStream manageableStream;

    @Test
    public void contextLoads() {
    }

}

那么Kafka团队就不会启动，所以上述问题就不会发生。

如果你想测试KafkaStreams的拓扑结构，请查看下面的链接

https://kafka.apache.org/11/documentation/streams/developer-guide/testing.html

我希望我的回答有帮助