Testing a @KafkaListener using Spring Embedded Kafka

Maybe someone will find this useful. I had a similar problem. Locally tests were running (some checks were performed within Awaitility.waitAtMost) but in the Jenkins pipeline, tests were failing.

The solution was, like already mentioned in the most voted answer, setting auto-offset-reset=earliest. When tests are running, you can check if you set the configuration properly by looking into test logs. Spring outputs configuration for both producer and consumer

You are probably sending the message before the consumer has been assigned the topic/partition. Set property...

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest

...it defaults to latest.

This is like using --from-beginning with the console consumer.

EDIT

Oh; you're not using boot's properties.

Add

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

EDIT2

BTW, you should probably also do a get(10L, TimeUnit.SECONDS) on the result of the template.send() (a Future<>) to assert that the send was successful.

EDIT3

To override the offset reset just for the test, you can do the same as what you did for the broker addresses:

@Value("${spring.kafka.consumer.auto-offset-reset:latest}")
private String reset;

...

    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, this.reset);

and

@TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
        "spring.kafka.consumer.auto-offset-reset=earliest"})

However, bear in mind that this property only applies the first time a group consumes. To always start at the end each time the app starts, you have to seek to the end during startup.

Also, I would recommend setting enable.auto.commit to false so that the container takes care of committing the offsets rather than just relying on the consumer client doing it on a time schedule.