title: How to Create Integration Test for Spring Kafka Listener
body: <p>I have one MicroService who sends to another microservice message that should consume. </p>
  
  <p>So, the kafka configs works, and everything works, but I need to create one intregration test for this code, and I have no idea how.</p>
  
  <p>My KafkaConsumer.Class anothed with component anottation:</p>
  
  <pre><code>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&lt;String, String&gt; 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&lt;Reaction&gt; 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");
      }
  
  }
  </code></pre>
  
  <p>My Integration Test is </p>
  
  <pre><code>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&lt;String, String&gt; consumer;
  
    @SuppressWarnings("SpringJavaAutowiringInspection")
    @Autowired
    public KafkaReactionMessageConsumerTest(KafkaReactionConsumerMessageComponent kafkaReactionConsumerMessageComponent,
                                          EmbeddedKafkaBroker embeddedKafkaBroker) {
      this.kafkaReactionConsumerMessageComponent = kafkaReactionConsumerMessageComponent;
      this.embeddedKafkaBroker = embeddedKafkaBroker;
  }
  
  @BeforeEach
  public void setUp() {
      Map&lt;String, Object&gt; configs = new HashMap&lt;&gt;(KafkaTestUtils.consumerProps("consumer", "true", embeddedKafkaBroker));
      consumer = new DefaultKafkaConsumerFactory&lt;&gt;(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&lt;String, Object&gt; configs = new HashMap&lt;&gt;(KafkaTestUtils.producerProps(embeddedKafkaBroker));
      Producer&lt;String, String&gt; producer = new DefaultKafkaProducerFactory&lt;&gt;(configs, new StringSerializer(), new StringSerializer()).createProducer();
  
  
      producer.send(new ProducerRecord&lt;&gt;(TOPIC, "1", reactionMessage.toString()));
      producer.flush();
  
      assertEquals(3, mongoTemplate.getCollection("reactions").countDocuments());
  }
  </code></pre>
  
  <p>The AbstractClass:</p>
  
  <pre><code> @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);
      }
  </code></pre>