1

我已经成功地使用 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("me@me.com")
                        .to(new String[]{"you@you.com"})
                        .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字段上添加 assertEquals，它在 JSON 中是一个电子邮件地址列表，由分隔，;并且在类中是一个字符串数组。因此，在EmailMessage课堂上，我的to字段注释为@JsonDeserializer指向拆分的自定义反序列化器;等。运行应用程序时一切正常，只有测试中断。

我尝试修改上面的代码以更改Producer<String, KafkaEmailMessageWrapper> producer为private Producer<String, String> 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)

编辑

让我重新表述这个问题，因为也许我有点混淆了。该应用程序的概述是：

  • 使用 JSON 消息
  • 将其序列化为KafkaEmailMessageWrapper
  • 发送电子邮件

这行得通，我向 Kafka 发送了一个 JSON，它被接收（上面的侦听器定义）并正确进行。我唯一的问题是在测试时。

我想在测试中模仿相同的情况，但是当我这样做时：

kafkaTemplate.sendDefault("key", "<json message>");

我得到一个：

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实例时：

kafkaTemplate.sendDefault("key", kafkaEmailMessageWrapper);

它可以工作，但是它不使用我@JsonDeserialize在课堂上指定的注释：

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

1 回答 1

0

您需要显示堆栈跟踪，但我猜因为您正在使用@SpringBootTest，真正的消费者（侦听器方法）也在获取消息，这就是您收到异常的原因。

目前尚不清楚您在此处测试的是什么（使用您的 test Consumer）对象。

一个更好的测试是，例如，将MailSender服务注入到您的侦听器 bean 中，并在测试用例中注入一个模拟MailSender来验证它是否按预期工作。

于 2019-06-05T12:54:59.857 回答