如何为TopologyTestDriver配置Input/OutputTestTopic的分区数？

完全支持为测试主题指定多分区，以此模拟生产环境的复杂场景，充分验证你的Kafka Streams处理逻辑。

具体实现方式 #

1. 创建测试主题时指定分区数
   通过TestTopicConfig构建器可以明确设置分区数量，在创建InputTestTopic时传入该配置即可：

   // 配置输入主题为3个分区
   TestTopicConfig inputTopicConfig = TestTopicConfig.builder()
       .partitions(3)
       .build();
   
   // 初始化TopologyTestDriver
   TopologyTestDriver testDriver = new TopologyTestDriver(yourTopology, streamsConfig);
   
   // 创建多分区输入测试主题
   InputTestTopic<String, String> multiPartitionInput = testDriver.createInputTopic(
       "your-input-topic",
       Serdes.String().serializer(),
       Serdes.String().serializer(),
       inputTopicConfig
   );
   
   // 对应的输出测试主题会自动匹配输入的分区配置
   OutputTestTopic<String, String> multiPartitionOutput = testDriver.createOutputTopic(
       "your-output-topic",
       Serdes.String().deserializer(),
       Serdes.String().deserializer()
   );
   

2. 向指定分区发送测试记录
   使用InputTestTopic的重载send方法，指定目标分区即可模拟不同分区的消息流入：

   // 往分区0发送消息
   multiPartitionInput.send("key-0", "value-0", 0);
   // 往分区1发送消息
   multiPartitionInput.send("key-1", "value-1", 1);
   // 往分区2发送消息
   multiPartitionInput.send("key-2", "value-2", 2);
   

3. 验证指定分区的输出结果
   通过OutputTestTopic的readRecordsFromPartition方法，可以精准读取目标分区的处理结果，方便针对多分区逻辑做断言：

   // 读取分区0的输出记录
   List<TestRecord<String, String>> partition0Output = multiPartitionOutput.readRecordsFromPartition(0);
   // 读取分区1的输出记录
   List<TestRecord<String, String>> partition1Output = multiPartitionOutput.readRecordsFromPartition(1);
   
   // 针对各分区结果做业务断言
   assertThat(partition0Output).hasSize(1);
   assertThat(partition0Output.get(0).value()).isEqualTo("processed-value-0");
   

这种方式可以完美模拟生产环境中的多分区场景，覆盖诸如按Key分区处理、跨分区状态交互等复杂逻辑的测试场景，确保你的Kafka Streams应用行为符合预期。

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