Spring Cloud Stream测试绑定器对KStream处理器无效问题求助

问题场景 #

我无法让Spring Cloud Stream测试绑定器适配基于KStream的doNothing函数，但它在非KStream的uppercase函数上可以正常工作（生产环境使用Kafka绑定器）。调试发现InputDestination和OutputDestination仅包含与uppercase相关的通道，没有doNothing的通道，执行doNothingTest时抛出空指针异常。

相关代码与配置 #

配置类代码 #

@Configuration
public class CPPNotificationConfiguration {
 
    @Bean
    public Function<KStream<byte[], byte[]>, KStream<byte[], byte[]>> doNothing() {
        return a -> a.flatMap((key, value) -> Collections.singleton(KeyValue.pair(key,value)));
    }

    @Bean
    public Function<String, String> uppercase() {
        return String::toUpperCase;
    }
}

绑定配置（application.yaml） #

spring:
  cloud.stream:
    function:
      definition: doNothing;uppercase
    bindings:
      doNothing-in-0:
        destination: doNothing-in-topic
      doNothing-out-0:
        destination: doNothing-out-topic
      uppercase-in-0:
        destination: uppercase-in-topic
      uppercase-out-0:
        destination: uppercase-out-topic

测试用例代码 #

@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
class CPPNotificationtest {
    @Autowired private InputDestination input;
    @Autowired private OutputDestination output;

    @Test
    void upperCaseTest() {
        input.send(new GenericMessage<>("asdfa"), "uppercase-in-topic");
        Message<byte[]> message = output.receive(100, "uppercase-out-topic");
        Assertions.assertArrayEquals("asdfa".toUpperCase().getBytes(), message.getPayload());
    }

    @Test
    void doNothingTest() {
        input.send(new GenericMessage<>("asdfa".getBytes()), "doNothing-in-topic");
        Message<byte[]> message = output.receive(100, "doNothing-out-topic");
        Assertions.assertArrayEquals("asdfa".getBytes(), message.getPayload());
    }
}

异常信息 #

java.lang.NullPointerException
    at org.springframework.cloud.stream.binder.test.InputDestination.send(InputDestination.java:89)
    at com.mokapos.cpp.notification.integration.UpperCaseTest.itemNotificationTest(UpperCaseTest.java:31)

Build.gradle依赖配置 #

implementation "org.springframework.boot:spring-boot-starter-web"
implementation "org.springframework.boot:spring-boot-starter-logging"
implementation 'org.springframework.boot:spring-boot-starter-test'
implementation "org.springframework.boot:spring-boot-starter-actuator"

implementation "org.springframework.cloud:spring-cloud-stream"
implementation "org.springframework.cloud:spring-cloud-stream-binder-kafka"
implementation "org.springframework.cloud:spring-cloud-stream-binder-kafka-streams"
testImplementation("org.springframework.cloud:spring-cloud-stream") {
        artifact {
            name = "spring-cloud-stream"
            extension = "jar"
            type ="test-jar"
            classifier = "test-binder"
        }
    }

问题原因 #

Spring Cloud Stream的测试绑定器（Test Channel Binder）仅支持标准的消息通道绑定，并不适配Kafka Streams绑定器。Kafka Streams函数依赖专门的Kafka Streams绑定器逻辑，测试绑定器无法识别和创建对应的KStream输入输出通道，因此InputDestination和OutputDestination中不会包含doNothing函数相关的通道，调用send时就会触发空指针异常。

解决方案 #

针对Kafka Streams函数的测试，需要使用专门的测试方式，推荐两种方案：

方案1：使用嵌入式Kafka进行测试 #

1. 添加嵌入式Kafka依赖到build.gradle：

testImplementation 'org.springframework.kafka:spring-kafka-test'

2. 改造测试用例，使用@EmbeddedKafka注解启动嵌入式Kafka，直接通过Kafka生产者/消费者发送和接收消息：

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"doNothing-in-topic", "doNothing-out-topic"})
class CPPNotificationtest {

    @Autowired
    private KafkaTemplate<byte[], byte[]> kafkaTemplate;

    @Autowired
    private ConsumerFactory<byte[], byte[]> consumerFactory;

    @Test
    void doNothingTest() throws InterruptedException {
        // 发送消息到输入主题
        kafkaTemplate.send("doNothing-in-topic", "asdfa".getBytes());

        // 创建消费者接收输出主题消息
        Consumer<byte[], byte[]> consumer = consumerFactory.createConsumer();
        consumer.subscribe(Collections.singleton("doNothing-out-topic"));
        ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(1000));

        // 断言结果
        Assertions.assertEquals(1, records.count());
        ConsumerRecord<byte[], byte[]> record = records.iterator().next();
        Assertions.assertArrayEquals("asdfa".getBytes(), record.value());

        consumer.close();
    }
}

方案2：使用Kafka Streams TestUtils（细粒度测试） #

如果需要更细粒度测试KStream逻辑，可以直接使用Kafka Streams提供的TestUtils，无需启动完整的Spring上下文：

class DoNothingFunctionTest {

    @Test
    void testDoNothingFunction() {
        // 初始化目标函数
        Function<KStream<byte[], byte[]>, KStream<byte[], byte[]>> doNothing = new CPPNotificationConfiguration().doNothing();

        // 创建测试用输入/输出主题
        TestInputTopic<byte[], byte[]> inputTopic = TestInputTopic.create("doNothing-in-topic", Serdes.ByteArray().serializer(), Serdes.ByteArray().serializer());
        TestOutputTopic<byte[], byte[]> outputTopic = TestOutputTopic.create("doNothing-out-topic", Serdes.ByteArray().deserializer(), Serdes.ByteArray().deserializer());

        // 处理输入并验证输出
        doNothing.apply(inputTopic.createKStream()).to(outputTopic);
        inputTopic.pipeInput("key".getBytes(), "asdfa".getBytes());

        Assertions.assertArrayEquals("asdfa".getBytes(), outputTopic.readValue());
    }
}

注意事项 #

• 测试绑定器仅适用于普通的Function<T,R>、Supplier<T>、Consumer<T>等基于消息通道的函数，不适用于KStream/KTable这类Kafka Streams特有的流式处理函数。
• 使用嵌入式Kafka时，Spring Cloud Stream默认会自动适配嵌入式Kafka地址，无需额外配置。

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