如何在使用模式注册表时对kafka streams dsl进行单元测试

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.test.TestRecord;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.Properties;

import static org.junit.Assert.assertEquals;

public class KafkaStreamsUnitTest {
    private TopologyTestDriver testDriver;
    private TestInputTopic<String, String> inputTopic;
    private TestOutputTopic<String, String> outputTopic;

    @Before
    public void setUp() {
        Properties props = new Properties();
        props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
        props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

        StreamsBuilder builder = new StreamsBuilder();
        // 构建拓扑结构
        builder.stream("input-topic").to("output-topic");

        Topology topology = builder.build();
        testDriver = new TopologyTestDriver(topology, props);

        inputTopic = testDriver.createInputTopic("input-topic", new StringSerializer(), new StringSerializer());
        outputTopic = testDriver.createOutputTopic("output-topic", new StringDeserializer(), new StringDeserializer());
    }

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

    @Test
    public void testKafkaStreams() {
        // 模拟输入数据
        inputTopic.pipeInput("key", "value");

        // 读取输出数据
        TestRecord<String, String> outputRecord = outputTopic.readRecord();

        // 验证输出结果
        assertEquals("key", outputRecord.getKey());
        assertEquals("value", outputRecord.getValue());
    }
}