Quarkus+Kafka Streams：GlobalKTable单元测试连接Bootstrap服务器失败

问题原因 #

GlobalKTable的设计逻辑是在Kafka Streams应用启动时，全量同步目标主题的所有数据到本地状态存储，确保每个实例拥有完整的全局视图。即便使用TopologyTestDriver，Quarkus集成的Kafka Streams扩展仍会触发GlobalKTable的初始状态恢复流程，尝试连接配置中的Kafka Broker拉取数据；而普通KTable仅基于测试输入的消息构建状态，不会触发全量恢复，因此不会出现连接问题。

解决方案 #

1. 自定义测试配置，阻止真实Broker连接

创建TopologyTestDriver时传入专属配置，禁用状态恢复并指定虚拟Broker地址，避免触发真实连接：

@BeforeEach
public void setup() throws IOException {
    Properties testProps = new Properties();
    // 测试用应用ID
    testProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-id-1");
    // 虚拟Broker地址，不会实际建立连接
    testProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    // 临时状态目录，避免污染本地环境
    testProps.put(StreamsConfig.STATE_DIR_CONFIG, 
        Files.createTempDirectory("kafka-streams-test").toAbsolutePath().toString());
    // 设置为at-most-once，禁用状态恢复和容错机制
    testProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.AT_MOST_ONCE);
    // 禁止自动重置偏移量，避免触发Broker数据拉取
    testProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");

    topologyTestDriver = new TopologyTestDriver(myTopology, testProps);

    // 注意：InputTopic序列化器需与GlobalKTable的Serde匹配
    inputTopic = topologyTestDriver.createInputTopic(
            "input-topic",
            new LongSerializer(), // 对应GlobalKTable的Long类型key
            new CustomAvroSerializer());
}

2. 修正测试数据的key类型

你的GlobalKTable使用Serdes.Long()作为key的Serde，但测试中误用StringSerializer传入字符串key，会导致状态存储无法匹配数据。调整测试输入的key为Long类型：

@Test
public void test() {
    // 传入Long类型key，与GlobalKTable的key类型一致
    inputTopic.pipeInput(10L, createValue());
    KeyValueStore<Long, CustomAvro> store = topologyTestDriver.getKeyValueStore(
            KafkaStreamsProperties.STATE_STORE);
    CustomAvro value = store.get(10L);
    assertEquals(10L, value.getSomeField()); // 替换为实际字段名
}

3. （可选）通过Quarkus测试配置统一覆盖

若使用Quarkus测试配置文件（src/test/resources/application-test.properties），可添加以下配置统一覆盖Kafka Streams设置，避免重复代码：

kafka-streams.bootstrap.servers=dummy:9092
kafka-streams.state.dir=${java.io.tmpdir}/kafka-streams-test
kafka-streams.processing.guarantee=at-most-once
kafka-streams.consumer.auto-offset-reset=none

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