Spring-kafka-test тестирование JSON-сообщения с настраиваемой десериализацией - PullRequest
Новая
       20

Spring-kafka-test тестирование JSON-сообщения с настраиваемой десериализацией

1 голос
/ 03 июня 2019

Я в основном успешно настроил тест с использованием 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());
    }

}

Это работает нормально, если я не добавлю assertEquals в поле to, которое в 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)

EDIT

Позвольте мне перефразировать проблему, потому что, возможно, я немного запутал вещи.Обзор приложения:

  • потребляет JSON-сообщение
  • сериализует его в KafkaEmailMessageWrapper
  • отправляет электронное письмо

Thisработает, я посылаю JSON в Kafka, он получен (определение слушателя выше) и работает правильно.Единственная проблема, с которой я сталкиваюсь - это тестирование.

Я хочу повторить ту же ситуацию в тесте, но когда я делаю это:

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

1 Ответ

0 голосов
/ 05 июня 2019

Вам нужно показать трассировку стека, но я предполагаю, что, поскольку вы используете @SpringBootTest, реальный потребитель (метод слушателя) также получает сообщение, поэтому вы получаете исключение.

Непонятно, что вы тестируете здесь (со своим тестом Consumer) объекта.

Лучшим тестом было бы, скажем, добавить сервис MailSender в ваш компонент-слушатель и в тестеcase вводит макет MailSender, чтобы убедиться, что все работает как положено.

...