EmbeddedKafka sending messages to consumer after delay in subsequent test

Question

I'm using EmbeddedKafka from kafka-spring-test, version 2.0.1.RELEASE (the latest). I have very simple tests which work correctly when running only one test. But whenever I run them one after another (so just running whole test class) then second one fails - consumer doest not receive any message.

public class KafkaControllerTest {

    private static final String FOO_TOPIC = "fooTopic";

    @Rule
    public KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, FOO_TOPIC);

    MessageConsumer msgConsumerFoo = mock(MessageConsumer.class);

    @Before
    public void before() {
        assertTrue(embeddedKafka.getBrokerAddresses().length == 1);
        KafkaController controller = new KafkaController(
                embeddedKafka.getBrokerAddress(0).toString(),
                new Consumer(FOO_TOPIC, msgConsumerFoo));
        MockMvcBuilders.standaloneSetup(controller).build();
    }

    @Test
    public void kafkaFirstTest() throws Exception {
        sendMessage(FOO_TOPIC, "foo message");

        verify(msgConsumerFoo).consume(any());
    }

    @Test
    public void kafkaSecondTest() throws Exception {
        sendMessage(FOO_TOPIC, "foo2 message");

        verify(msgConsumerFoo).consume(any());
    }

    void sendMessage(String topic, String notification) throws ExecutionException, InterruptedException {
        Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka);
        Producer<Integer, String> producer = new DefaultKafkaProducerFactory<Integer, String>(props).createProducer();
        producer.send(new ProducerRecord<>(topic, notification)).get();
    }
}

And code of tested class:

public class KafkaController {

    public KafkaController(String brokerAddress, Consumer... consumers) {
        for (Consumer consumer : consumers) {
            addTopicListener(brokerAddress, consumer.topic, consumer.messageConsumer);
        }
    }

    private void addTopicListener(String brokerAddress, String topic, MessageConsumer consumer) {
        HashMap<String, Object> consumerConfig = new HashMap<>();
        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerAddress);
        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
        DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
                consumerConfig,
                new StringDeserializer(),
                new StringDeserializer());

        ContainerProperties containerProperties = new ContainerProperties(topic);
        containerProperties.setMessageListener((MessageListener<String, String>) data -> {
            consumer.consume(data.value());
        });
        ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProperties);
        container.start();
    }

    public interface MessageConsumer {
        void consume(String message);
    }

    public static class Consumer {
        private final String topic;
        private final MessageConsumer messageConsumer;

        public Consumer(String topic, MessageConsumer messageConsumer) {
            this.topic = topic;
            this.messageConsumer = messageConsumer;
        }
    }
}

I believe the problem is caused by KafkaEmbedded as consumers tested with local Kafka instance work correctly.

Is there something I'm missing? I didn't find any consumer property that could be helpful.

Kafka-spring-test doc say:

It is generally recommended to use the rule as a @ClassRule to avoid starting/stopping the broker between tests (and use a different topic for each test).

However, I tried both making KafkaEmbedded as @ClassRule and using different topic in tests and still nothing.

This issue has something to do with asynchronicity, because adding delays in second test helps:

@Test
public void kafkaSecondTest() throws Exception {
    Thread.sleep(1000);
    sendMessage(FOO_TOPIC, "foo2 message");
    Thread.sleep(1000);

    verify(msgConsumerFoo).consume(any());
}

Yes, somehow it works only when Thread.sleep(1000) is added both before and after sending message.

So how can I check if KafkaEmbedded or some other component is ready to send/consume messages?

Answer 1

You probably need to set auto.offset.reset=earliest (ConsumerConfig.AUTO_OFFSET_RESET_CONFIG). Otherwise you might send the message before the container completely starts.

Of course, if you use the same topic, the second consumer should get 2 messages.

Also, I would recommend stopping the container at the end of each test. Debug logging should help too.

EDIT

An alternative is to use ContainerTestUtils.waitForAssignment() before sending the message in the test; we do that a lot in the framework tests.