java - 测试 spring 云流服务的问题

Question

我在测试我的 spring 云流服务(写入 Kafka 流)时遇到问题。它基于以下 baeldung introduction

这里是服务代码(具体省略)

@Service
@EnableBinding(Source.class)
public class KafkaWriterService {

    @SendTo(Source.OUTPUT)
    public String write(String str) {
        return str;
    }

}

下面是测试

@SpringBootTest
@RunWith(SpringRunner.class)
public class KafkaWriterServiceTest {

    @Autowired
    private Source source;

    @Autowired
    private MessageCollector collector;

    @Autowired
    private KafkaWriterService service;

    @Test
    public void testMessages() {
        BlockingQueue<Message<?>> messages = collector.forChannel(source.output());
        service.write("FooBar");

        Object payload = messages.poll().getPayload();
        System.out.println(payload);
    }

非常简单，但是在执行测试时我得到一个 NullPointerException，因为 poll 返回一个 Null。

知道问题出在哪里吗？

谢谢!

Answer 1

更新:

如果您只是在寻找 Kafka 生产者示例，请对您的服务进行以下更改:

@Service
@EnableBinding({Source.class})
public class KafkaWriterService {

    private final MessageChannel output;

    public KafkaWriterService(@Qualifier("output") MessageChannel output) {
        this.output = output;
    }

    public void write(String str) {
        output.send(MessageBuilder.withPayload(str).build());
    }
}

---

@SendTo 需要通过 @StreamListener 的消息源。它用于转换输入消息。

在您的测试程序中，没有生成任何消息。

将您的代码更改为以下内容，以便在您的测试中生成一条消息:

@Service
@EnableBinding({Source.class, Sink.class})
public class KafkaWriterService {

    @StreamListener(Sink.INPUT)
    @SendTo(Source.OUTPUT)
    public String write(String str) {
        return "Transformed: " + str;
    }
}

@SpringBootTest
@RunWith(SpringRunner.class)
public class KafkaWriterServiceTest {

    @Autowired
    private Sink sink;

    @Autowired
    private Source source;

    @Autowired
    private MessageCollector collector;

    @Autowired
    private KafkaWriterService service;

    @Test
    public void testMessages() {
        BlockingQueue<Message<?>> messages = collector.forChannel(source.output());
        sink.input().send(MessageBuilder.withPayload("Hello World!").build());

        Object payload = messages.poll().getPayload();
        System.out.println(payload);
    }
}