Spring-kafka-test使用自定义反序列化测试JSON消息

Question

我基本上成功地使用EmbeddedKafka设置了一个测试，它生成并使用一条消息。下面是测试的工作版本：

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka
public class KafkaFlowTest {

    @Autowired
    EmbeddedKafkaBroker broker;

    public static final String EMAIL_TOPIC = "email-service";

    static {
        System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY, "spring.kafka.bootstrap-servers");
    }

    private Consumer<String, KafkaEmailMessageWrapper> consumer;
    private Producer<String, KafkaEmailMessageWrapper> producer;

    @Before
    public void setUp() {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "true", broker);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        Map<String, Object> producerProps = KafkaTestUtils.producerProps(broker);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        DefaultKafkaConsumerFactory<String, KafkaEmailMessageWrapper> cf =
                new DefaultKafkaConsumerFactory<>(
                        consumerProps,
                        new StringDeserializer(),
                        new JsonDeserializer<>(KafkaEmailMessageWrapper.class)
                );
        consumer = cf.createConsumer();
        consumer.subscribe(Collections.singleton(EMAIL_TOPIC));
        consumer.poll(Duration.ZERO);

        DefaultKafkaProducerFactory<String, KafkaEmailMessageWrapper> pf =
                new DefaultKafkaProducerFactory<>(producerProps);
        producer = pf.createProducer();
    }

    @Test
    public void testNoErrorMessageFlow() {
        KafkaEmailMessageWrapper wrapperMessage = KafkaEmailMessageWrapper.builder()
                .emailRequest(
                    EmailMessage.builder()
                        .body("body")
                        .from("[e-mail]")
                        .to(new String[]{"[e-mail]"})
                        .subject("hey!")
                        .build()
                )
                .build();

        producer.send(new ProducerRecord<>(EMAIL_TOPIC, "whatever", wrapperMessage));
        producer.flush();

        ConsumerRecord<String, KafkaEmailMessageWrapper> record = KafkaTestUtils.getSingleRecord(consumer, EMAIL_TOPIC);
        KafkaEmailMessageWrapper receivedMessage = record.value();
        assertEquals(wrapperMessage.getEmailRequest().getSubject(), receivedMessage.getEmailRequest().getSubject());
    }

}

除非在to字段上添加assertequal，否则这是正常的，在JSON中，该字段是一个电子邮件地址列表，由分隔；在类中是一个字符串数组。因此，在EmailMessage类中，我有to字段，该字段用@JSONDeserializer注释，该字段指向沿着拆分的自定义反序列化器；等。

我试图修改上面的代码，将生产者 producer 更改为私有生产者 producer ，然后在测试中发送JSON而不是kafkaeMailMessageWrapper实例，但我得到了一个异常：

java.lang.ClassCastException: java.lang.String cannot be cast to com.test.model.KafkaEmailMessageWrapper

因此，由于某种原因，从JSON字符串到模型的反序列化似乎没有在本测试场景中发生。我希望测试尽可能接近真实的用例，因此它应该产生一个字符串消息，然后对我的模型类执行反序列化。不确定为什么这不发生，任何帮助理解为什么这将是感激的！

为了完成，这是消息侦听器的定义：

@KafkaListener(topics = "email-service")
public void receive(@Payload KafkaEmailMessageWrapper message)

编辑

null

kafkatemplate.senddefault（“key”,“ ”）；

我得到一个：

Cannot handle message; nested exception is  org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.test.model.KafkaEmailMessageWrapper] for GenericMessage

但是，当我发送KafkaEmailMessageWrapper实例时：

@JsonDeserialize(using = SemicolonArrayDeserializer.class)
private String[] to;
@JsonDeserialize(using = SemicolonArrayDeserializer.class)
private String[] cc;

因此，当发送实例时，to和cc字段总是空的，因为字段中已经是字符串数组，所以沿着；拆分的反序列化逻辑不会起任何作用。这是反序列化方法：

public String[] deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
    ObjectMapper mapper = (ObjectMapper) p.getCodec();
    JsonNode node = mapper.readTree(p);
    return node.asText().split(";");
}

Answer 1

您需要显示堆栈跟踪，但我猜测，因为您使用的是@springboottest，所以真正的使用者（侦听器方法）也会得到消息，这就是为什么您会得到异常的原因。

不清楚您在这里（使用您的testconsumer)对象测试的是什么。

更好的测试是将mailsender服务注入到您的侦听器bean中，并在测试用例中注入一个模拟mailsender来验证它都能正常工作。