如何使用Spring Kafka测试Kafka Streams应用？

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = "inputTopic", controlledShutdown = true)
public class KafkaStreamsApplicationTests {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private StreamsBuilderFactoryBean streamsBuilderFactoryBean;

    private TopologyTestDriver testDriver;

    @Before
    public void setup() {
        Properties config = streamsBuilderFactoryBean.getStreamsConfiguration();
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-application");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        testDriver = new TopologyTestDriver(streamsBuilderFactoryBean.getTopology(), config);
    }

    @After
    public void cleanup() {
        testDriver.close();
    }

    @Test
    public void testKafkaStreamsApplication() {
        // 发送输入数据到Kafka主题
        kafkaTemplate.send("inputTopic", "key", "value");

        // 从输出主题中读取结果
        TestOutputTopic<String, String> outputTopic = testDriver.createOutputTopic("outputTopic", new StringDeserializer(), new StringDeserializer());
        KeyValue<String, String> result = outputTopic.readKeyValue();

        // 验证输出结果是否符合预期
        assertThat(result.key()).isEqualTo("key");
        assertThat(result.value()).isEqualTo("processed value");
    }
}