Testing a @KafkaListener using Spring Embedded Kafka : <p>I am trying to write a unit test for a Kafka listener that I am developing using Spring Boot 2.x. Being a unit test, I don't want to start up a full Kafka server an instance of Zookeeper. So, I decided to use Spring Embedded Kafka.</p>

<p>The definition of my listener is very basic.</p>

<pre><code>@Component
public class Listener {
    private final CountDownLatch latch;

    @Autowired
    public Listener(CountDownLatch latch) {
        this.latch = latch;
    }

    @KafkaListener(topics = "sample-topic")
    public void listen(String message) {
        latch.countDown();
    }
}
</code></pre>

<p>Also the test, that verifies the <code>latch</code> counter to be equal to zero after receiving a message, is very easy.</p>

<pre><code>@RunWith(SpringRunner.class)
@SpringBootTest
@DirtiesContext
@EmbeddedKafka(topics = { "sample-topic" })
@TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}" })
public class ListenerTest {

    @Autowired
    private KafkaEmbedded embeddedKafka;

    @Autowired
    private CountDownLatch latch;

    private KafkaTemplate&lt;Integer, String&gt; producer;

    @Before
    public void setUp() {
        this.producer = buildKafkaTemplate();
        this.producer.setDefaultTopic("sample-topic");
    }

    private KafkaTemplate&lt;Integer, String&gt; buildKafkaTemplate() {
        Map&lt;String, Object&gt; senderProps = KafkaTestUtils.producerProps(embeddedKafka);
        ProducerFactory&lt;Integer, String&gt; pf = new DefaultKafkaProducerFactory&lt;&gt;(senderProps);
        return new KafkaTemplate&lt;&gt;(pf);
    }

    @Test
    public void listenerShouldConsumeMessages() throws InterruptedException {
        // Given
        producer.sendDefault(1, "Hello world");
        // Then
        assertThat(latch.await(10L, TimeUnit.SECONDS)).isTrue();
    }
}
</code></pre>

<p>Unfortunately, the test fails and I cannot understand why. Is it possible to use an instance of <code>KafkaEmbedded</code> to test a method marked with the annotation <code>@KafkaListener</code>?</p>

<p>All the code is shared in my GitHub repository <a href="https://github.com/rcardin/kafka-listener" rel="noreferrer">kafka-listener</a>.</p>

<p>Thanks to all.</p>
