如何使用Kafka Streams对Spring Cloud Stream进行单元测试

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.test.ConsumerRecordFactory;
import org.apache.kafka.streams.test.OutputVerifier;
import org.apache.kafka.streams.test.TestRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration.TestChannelBinder;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import static org.assertj.core.api.Assertions.assertThat;

public class KafkaStreamsSpringCloudStreamUnitTest {

    private AnnotationConfigApplicationContext context;
    private InputDestination input;
    private OutputDestination output;

    @BeforeEach
    public void setup() {
        context = new AnnotationConfigApplicationContext();
        context.register(TestChannelBinderConfiguration.class);
        context.refresh();

        TestChannelBinder binder = context.getBean(TestChannelBinder.class);
        input = binder.createInput("input");
        output = binder.createOutput("output");
    }

    @AfterEach
    public void cleanup() {
        context.close();
    }

    @Test
    public void testKafkaStreamsSpringCloudStreamIntegration() {
        // 准备测试数据
        String inputTopic = "input";
        String outputTopic = "output";
        String inputValue = "test message";
        TestRecord<String, String> testRecord = new TestRecord<>(inputValue);

        // 配置Kafka Streams属性
        Properties properties = new Properties();
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-application");
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 创建Kafka Streams拓扑
        StreamsBuilder builder = new StreamsBuilder();
        builder.stream(inputTopic).to(outputTopic);

        // 启动Kafka Streams应用程序
        Topology topology = builder.build();
        KafkaStreams kafkaStreams = new KafkaStreams(topology, properties);
        kafkaStreams.start();

        // 发送测试数据到输入主题
        input.send(testRecord);

        // 验证输出结果
        Message<byte[]> message = output.receive();
        String actualValue = new String(message.getPayload());
        assertThat(actualValue).isEqualTo(inputValue);

        // 清理测试环境
        kafkaStreams.close();
        input.clear();
        output.clear();
    }
}