如何使用spring-boot EmbeddedKafka进行KStream拓扑的集成测试？

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

@Configuration
public class KafkaTestConfig {

    @Bean
    public EmbeddedKafkaBroker embeddedKafkaBroker() {
        return new EmbeddedKafkaBroker(1, true, "topic1", "topic2")
                .brokerProperty("listeners", "PLAINTEXT://localhost:9092")
                .brokerProperty("auto.create.topics.enable", "false");
    }
}

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka
@Import(KafkaTestConfig.class)
public class KafkaStreamIntegrationTest {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafkaBroker;

    @Test
    public void testKStreamTopology() throws Exception {
        // 创建KafkaProducer，发送测试数据到输入topic
        Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafkaBroker);
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
        producer.send(new ProducerRecord<>("input-topic", "key", "value"));
        producer.flush();

        // 创建KafkaConsumer，订阅输出topic，接收处理后的数据
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("test-group", "false", embeddedKafkaBroker);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Collections.singleton("output-topic"));

        // 等待一段时间，确保Kafka消息被处理
        Thread.sleep(5000);

        // 检查输出topic中的数据
        ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer);
        assertThat(records.count()).isEqualTo(1);
        ConsumerRecord<String, String> record = records.iterator().next();
        assertThat(record.key()).isEqualTo("key");
        assertThat(record.value()).isEqualTo("processed-value");
    }
}