无法使用 Kafka 测试在 Spring Cloud Stream 中读取 JSON

Question

我有一个简单的应用程序，它使用 spring cloud stream 和 kafka 接收消息并写入另一个主题。我正在尝试根据一些教程编写单元测试，但在发送消息时出现错误。根据消息，它无法读取 JSON。

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class ApplicationTests {

    @Autowired
    private Processor processor;

    @Autowired
    private MessageCollector messageCollector;


    @Test
    @SuppressWarnings("unchecked")
    public void testWriting() {

        MyRecord myRecord= getDummyRecord(); // getting dummy MyRecord object

        Message<MyRecord> message = MessageBuilder.withPayload(myRecord).setHeader(KafkaHeaders.MESSAGE_KEY, getDummyKey()).build();
        processor.input().send(message);

        Message<MyRecord> receivedMessage = (Message<MyRecord>) messageCollector.forChannel(processor.output()).poll();


    }

}

我得到的错误是：

无法读取 JSON：无法识别的令牌“WeatherRecordRaw”：期待（“true”、“false”或“null”） 在 [Source: (String)"MyRecord(....)

我该如何解决这个问题，我在应用中没有问题，只有在测试中。

Answer 1

您正在直接发送到输入通道； binder 从源（RabbitMQ、Kafka 等）获取数据并将其发送到通道进行转换。

输入通道期望得到 JSON，而不是 POJO；然后它将 JSON 转换回 POJO。

使用 Jackson ObjectMapper 将 POJO 转换为包含 JSON 表示的 byte[]。