Spring Kafka Unit Tests Triggers the listener, but the method cannot get the message using consumer.poll

Question

We are using spring-kafka-test-2.2.8-RELEASE. When I use the template to send the message, it triggers the listener correctly, but I can't get the message content in the consumer.poll. If i instantiate the KafkaTemplate without "wiring" it in a class attribute and Instantiate it based on a producer factory, it sends the message, but does not trigger the @KafkaListener, only work if I setup a Message Listener inside the @Test Method. I need to trigger the kafka listener and realize which Topic will be called next("sucess" topic when executed without errors, and "errorTopic" the listener throws an Exception) and the message content.

    @RunWith(SpringRunner.class)
    @SpringBootTest
    @EmbeddedKafka(partitions = 1, topics = { "tp-in-gco-mao-notasfiscais" })
    public class InvoicingServiceTest {

         @Autowired
         private NFKafkaListener nfKafkaListener;

         @ClassRule
         public static EmbeddedKafkaRule broker = new EmbeddedKafkaRule(1, false, "tp-in-gco-mao- 
         notasfiscais");

         @Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
         private String brokerAddresses;

         @Autowired
         private KafkaTemplate<Object, Object> template;

         @BeforeClass
         public static void setup() {
                System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY,
              "spring.kafka.bootstrap-servers");
         }

         @Test
         public void testTemplate() throws Exception {
               NFServiceTest nfServiceTest = spy(new NFServiceTest());

               nfKafkaListener.setNfServiceClient(nfServiceTest);
               Map<String, Object> consumerProps =  KafkaTestUtils.consumerProps("teste9", "false", broker.getEmbeddedKafka());
               consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
               consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
               consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, InvoiceDeserializer.class);
               consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

               DefaultKafkaConsumerFactory<Integer, Object> cf = new DefaultKafkaConsumerFactory<Integer, Object>(
            consumerProps);

           Consumer<Integer, Object> consumer = cf.createConsumer();

           broker.getEmbeddedKafka().consumeFromAnEmbeddedTopic(consumer, "tp-in-gco-mao-notasfiscais");

           ZfifNfMao zf = new ZfifNfMao();
           zf.setItItensnf(new Zfietb011());

           Zfietb011 zfietb011 = new Zfietb011();
           Zfie011 zfie011 = new Zfie011();
           zfie011.setMatkl("TESTE");
           zfietb011.getItem().add(zfie011);
           zf.setItItensnf(zfietb011);

           template.send("tp-in-gco-mao-notasfiscais", zf);
           List<ConsumerRecord<Integer, Object>> received = new ArrayList<>();
           int n = 0;
           while (received.size() < 1 && n++ < 10) {
                ConsumerRecords<Integer, Object> records1 = consumer.poll(Duration.ofSeconds(10));
                //records1  is always empty
                if (!records1.isEmpty()) {
                    records1.forEach(rec -> received.add(rec));
                }
           }

           assertThat(received).extracting(rec -> {
               ZfifNfMao zfifNfMaoRdesponse = (ZfifNfMao) rec.value();
               return zfifNfMaoRdesponse.getItItensnf().getItem().get(0).getMatkl();
            }).contains("TESTE");
            broker.getEmbeddedKafka().getKafkaServers().forEach(b -> b.shutdown());
            broker.getEmbeddedKafka().getKafkaServers().forEach(b -> b.awaitShutdown());
            consumer.close();
        }

        public static class NFServiceTest implements INFServiceClient {
            CountDownLatch latch = new CountDownLatch(1);

            @Override
            public ZfifNfMaoResponse enviarSap(ZfifNfMao zfifNfMao) {
                ZfifNfMaoResponse zfifNfMaoResponse = new ZfifNfMaoResponse();
                zfifNfMaoResponse.setItItensnf(new Zfietb011());

                Zfietb011 zfietb011 = new Zfietb011();
                Zfie011 zfie011 = new Zfie011();
                zfie011.setMatkl("TESTE");
                zfietb011.getItem().add(zfie011);
                zfifNfMaoResponse.setItItensnf(zfietb011);
                return zfifNfMaoResponse;
            }
        }
    }     

Answer 1

I've followed your advice but it keeps triggering the listener, but consumer.poll does not capture the topic content.

@RunWith(SpringRunner.class)
@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = { "tp-in-gco-mao-notasfiscais" })
public class InvoicingServiceTest {

     @Autowired
     private NFKafkaListener nfKafkaListener;

 @Autowired
 public EmbeddedKafkaBroker broker;

 @Autowired
 private KafkaTemplate<Object, Object> template;

 @BeforeClass
 public static void setup() {
        System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY,
      "spring.kafka.bootstrap-servers");
 }

 @Test
 public void testTemplate() throws Exception {
       NFServiceTest nfServiceTest = spy(new NFServiceTest());

       nfKafkaListener.setNfServiceClient(nfServiceTest);
       Map<String, Object> consumerProps =  KafkaTestUtils.consumerProps("teste9", "false", broker);
       consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
       consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
       consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, InvoiceDeserializer.class);

       DefaultKafkaConsumerFactory<Integer, Object> cf = new DefaultKafkaConsumerFactory<Integer, Object>(
    consumerProps);

   Consumer<Integer, Object> consumer = cf.createConsumer();

   broker.consumeFromAnEmbeddedTopic(consumer, "tp-in-gco-mao-notasfiscais");

   ZfifNfMao zf = new ZfifNfMao();
   zf.setItItensnf(new Zfietb011());

   Zfietb011 zfietb011 = new Zfietb011();
   Zfie011 zfie011 = new Zfie011();
   zfie011.setMatkl("TESTE");
   zfietb011.getItem().add(zfie011);
   zf.setItItensnf(zfietb011);

   template.send("tp-in-gco-mao-notasfiscais", zf);
   List<ConsumerRecord<Integer, Object>> received = new ArrayList<>();
   int n = 0;
   while (received.size() < 1 && n++ < 10) {
        ConsumerRecords<Integer, Object> records1 = consumer.poll(Duration.ofSeconds(10));
        //records1  is always empty
        if (!records1.isEmpty()) {
            records1.forEach(rec -> received.add(rec));
        }
   }

   assertThat(received).extracting(rec -> {
       ZfifNfMao zfifNfMaoRdesponse = (ZfifNfMao) rec.value();
       return zfifNfMaoRdesponse.getItItensnf().getItem().get(0).getMatkl();
    }).contains("TESTE");
    broker.getKafkaServers().forEach(b -> b.shutdown());
    broker.getKafkaServers().forEach(b -> b.awaitShutdown());
    consumer.close();
}

public static class NFServiceTest implements INFServiceClient {
    CountDownLatch latch = new CountDownLatch(1);

    @Override
    public ZfifNfMaoResponse enviarSap(ZfifNfMao zfifNfMao) {
        ZfifNfMaoResponse zfifNfMaoResponse = new ZfifNfMaoResponse();
        zfifNfMaoResponse.setItItensnf(new Zfietb011());

        Zfietb011 zfietb011 = new Zfietb011();
        Zfie011 zfie011 = new Zfie011();
        zfie011.setMatkl("TESTE");
        zfietb011.getItem().add(zfie011);
        zfifNfMaoResponse.setItItensnf(zfietb011);
        return zfifNfMaoResponse;
    }
}
} 

Answer 2

You have two brokers; one created by @EmbeddedKafka and one created by the @ClassRule.

Use one or the other; preferably the @EmbeddedKafka and simply @Autowired the broker instance.

I am guessing the consumers are listening to different brokers; you can confirm that by looking at the INFO logs put out by the consumer config.