Spring Cloud Stream Kafka 消费者测试

Question

我正在尝试按照 GitHub a link 的建议设置测试

        Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
    DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
    try {
        KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
        template.setDefaultTopic("words");
        template.sendDefault("foobar");

    --> ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer, "output");
        log.debug(cr);
    }
    finally {
        pf.destroy();
    }

StreamProcessor 设置为

@StreamListener
    @SendTo("output")
    public KStream<?, WordCount> process(@Input("input") KStream<Object, String> input) {

        return input.map((key, value) -> new KeyValue<>(value, new WordCount(value, 10, new Date(), new Date())));
    }

--> 由于@Streamprocessor 具有@SendTo("output") ，因此行从不消耗我认为应该在主题“输出”上的消息

• 我希望能够测试流处理的消息。

Answer 1

您需要从您的output 绑定的实际主题中消费。 你有spring.cloud.stream.bindings.output.destination的配置吗？那应该是您需要使用的值。如果您没有设置，默认值将与绑定相同 - 在这种情况下为 output。