Как создать интеграционный тест для Spring Kafka Listener

Question

У меня есть один MicroService, который отправляет другому сообщению MicroService, которое должно потреблять.

Итак, конфиги kafka работают, и все работает, но мне нужно создать один тест на интеграцию для этого кода, и я понятия не имею, как.

Мой KafkaConsumer.Class привязан к анотации компонентов :

private static final Logger logger = LoggerFactory.getLogger(KafkaReactionConsumerMessageComponent.class);

private final ReactionsService reactionsService;

public KafkaReactionConsumerMessageComponent(ReactionsService reactionsService) {
    this.reactionsService = reactionsService;
}

   @KafkaListener(topics = "reaction-topic", clientIdPrefix = "string", groupId = "magpie-trending")
   public void consumingReactionMessages(ConsumerRecord<String, String> cr,
                                      @Payload String payload){
    logger.info("[JSON] received Payload: {}", payload);

    try {
        ObjectMapper mapper = new ObjectMapper();
        ReactionMessage message = mapper.readValue(payload, ReactionMessage.class);
        if(StringUtils.equals("unloved", message.getReactionType())) {
            reactionsService.deleteReactionsByUserIdAndPostId(message.getPost().getPostId(), message.getUser().getUserId());
            logger.info("Deleted reactions from database with postId: {} and userId: {}", message.getPost().getPostId(), message.getUser().getUserId());
        } else {
            List<Reaction> reactions = creatReactions(message).stream()
                    .map(reactionsService::insertReaction).collect(Collectors.toList());
            logger.info("Added reactions to database: {}", reactions);
        }
    } catch (Exception e){
        logger.error("Cannot Deserialize payload to ReactionMessage");
    }

}

Мой интеграционный тест -

private static final String TOPIC = "reaction-topic";
  private final Logger logger = LoggerFactory.getLogger(KafkaDeletePostConsumerMessageComponent.class);

  private final KafkaReactionConsumerMessageComponent kafkaReactionConsumerMessageComponent;
  private final EmbeddedKafkaBroker embeddedKafkaBroker;

  private Consumer<String, String> consumer;

  @SuppressWarnings("SpringJavaAutowiringInspection")
  @Autowired
  public KafkaReactionMessageConsumerTest(KafkaReactionConsumerMessageComponent kafkaReactionConsumerMessageComponent,
                                        EmbeddedKafkaBroker embeddedKafkaBroker) {
    this.kafkaReactionConsumerMessageComponent = kafkaReactionConsumerMessageComponent;
    this.embeddedKafkaBroker = embeddedKafkaBroker;
}

@BeforeEach
public void setUp() {
    Map<String, Object> configs = new HashMap<>(KafkaTestUtils.consumerProps("consumer", "true", embeddedKafkaBroker));
    consumer = new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new StringDeserializer()).createConsumer();
    consumer.subscribe(Collections.singleton(TOPIC));
    consumer.poll(Duration.ZERO);
}

@AfterEach
public void tearDown() {
    consumer.close();
}

@Test
public void shoudlConsumeAndInsertInDatabaseReactionDomain() {
    ReactionMessage reactionMessage = new ReactionMessage(new PostMessage("1", Set.of("a", "b", "c")),
            new UserMessage("2"), LocalDateTime.now().toString(), "loved");

    Map<String, Object> configs = new HashMap<>(KafkaTestUtils.producerProps(embeddedKafkaBroker));
    Producer<String, String> producer = new DefaultKafkaProducerFactory<>(configs, new StringSerializer(), new StringSerializer()).createProducer();


    producer.send(new ProducerRecord<>(TOPIC, "1", reactionMessage.toString()));
    producer.flush();

    assertEquals(3, mongoTemplate.getCollection("reactions").countDocuments());
}

AbstractClass:

 @ExtendWith(SpringExtension.class)
  @SpringBootTest
  @AutoConfigureMockMvc
  @EmbeddedKafka(brokerProperties={
        "log.dir=out/embedded-kafka"
  })
  public abstract class AbstractMongoEmbeddedTest {

    @Autowired
    private static MongodExecutable mongodExecutable;

    @Autowired
    protected MongoTemplate mongoTemplate;

    @BeforeEach
    private void dropPostCollection(){
       mongoTemplate.dropCollection(Reaction.class);
    }

Answer 1

Поскольку вы используете встроенный брокер Kafka, вы можете просто производить / потреблять нужные топи c (s) из вашего интеграционного теста.

Потребление

Потребление может быть выполнено с помощью простого правила jUnit. Правило, служащее этой цели, можно найти здесь . Не стесняйтесь использовать его.

Вы можете использовать его для подтверждения потребляемых сообщений:

assertThat(kafkaConsumeMessagesRule.pollMessage()).isEqualTo("your-expected-message");

Производство

Для создания сообщений вы можете просто подключить org.springframework.kafka.core.KafkaTemplate в своем интеграционном тесте и отправлять сообщения на указанный номер c.