春云流Kafka消费测试

Question

我正试图按照GitHub的建议设置测试

        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())));
    }

-->line从不使用在我看来应该在主题“output”上的消息，因为@StreamProcessor有@Sendto（“output”）

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

Answer 1

您需要使用输出绑定到的实际主题。您有spring.cloud.stream.bindings.output.destination的配置吗？这应该是您需要使用的值。如果不设置，默认值将与绑定相同-在本例中输出。