JUnit执行完毕后Spring Boot无法停止，嵌入式Kafka问题求助

我来帮你梳理下这个问题的解决思路，亲测有效的几个方案：

1. 手动关闭Kafka Streams实例（核心解决线程残留） #

Kafka Streams的StreamThread是后台守护线程，测试结束后如果不主动关闭，Spring上下文无法正常销毁。你可以在测试类里注入KafkaStreams对象，然后在测试结束后调用关闭方法：

@Autowired
private KafkaStreams kafkaStreams;

@After
public void tearDownKafkaStreams() {
    // 等待5秒确保流完全关闭，时间可根据你的拓扑复杂度调整
    kafkaStreams.close(Duration.ofSeconds(5));
}

这样能主动终止所有后台流线程，让Spring顺利销毁上下文。

2. 修复嵌入式Kafka的Broker连接配置（解决节点-1错误） #

控制台里的Connection to node -1 could not be established错误，本质是你的Kafka Streams找不到正确的嵌入式Broker地址。嵌入式Kafka默认会用随机端口，不能硬编码localhost:9092，要动态获取Broker地址：

@Autowired
private EmbeddedKafkaBroker embeddedKafkaBroker;

@Bean
public KafkaStreams kafkaStreams() {
    Properties streamsProps = new Properties();
    // 动态获取嵌入式Kafka的Broker地址
    streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafkaBroker.getBrokersAsString());
    // 其他Streams配置（比如application.id、默认序列化器等）
    streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app-id");
    streamsProps.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    streamsProps.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    Topology customTopology = buildYourCustomTopology();
    return new KafkaStreams(customTopology, streamsProps);
}

同时要确保测试类上添加了@EmbeddedKafka注解，指定测试需要的Topic：

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(topics = {"input-topic", "output-topic"}, partitions = 1)
public class YourStreamTest {
    // 测试代码
}

3. 调整@DirtiesContext的生效时机 #

之前用@DirtiesContext没生效，大概率是默认的生效时机不对。默认@DirtiesContext是在整个测试类执行完后才销毁上下文，改成每个测试方法执行后就销毁：

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(topics = {"input-topic", "output-topic"})
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
public class YourStreamTest {
    // 测试代码
}

这样能避免测试方法之间的线程残留问题。

4. 添加JVM关闭钩子作为兜底 #

可以在创建KafkaStreams实例时添加关闭钩子，确保JVM退出时能强制关闭流：

KafkaStreams streams = new KafkaStreams(customTopology, streamsProps);
// 添加关闭钩子
Runtime.getRuntime().addShutdownHook(new Thread(() -> streams.close(Duration.ofSeconds(3))));
streams.start();

这个作为兜底方案，配合前面的手动关闭，基本能解决线程残留问题。

内容的提问来源于stack exchange，提问作者Silvio Papa