Creating a Kafka Listener Using the Consumer API

1. Overview

In this tutorial, we’ll learn how to create a Kafka listener and consume messages from a topic using Kafka’s Consumer API. After that, we’ll test our implementation using the Producer API and Testcontainers.

We’ll be focusing on setting up a KafkaConsumer without relying on Spring Boot modules.

2. Create a Custom Kafka Listener

Our custom listener will internally use the Producer and Consumer APIs from the kafka-clients library. Let’s start by adding this dependency to our pom.xml file:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version>
</dependency>

For the code examples in this article, we’ll create a CustomKafkaListener class that will listen to the topic named “baeldung.articles.published“. Internally, our class will wrap around a KafkaConsumer and leverage it to subscribe to the topic:

class CustomKafkaListener {
    private final String topic;
    private final KafkaConsumer<String, String> consumer;

    // ...
}

2.1. Create a KafkaConsumer

To create a KafkaConsumer, we need to supply a valid configuration via a Properties object. Let’s create a simple consumer that can be used as a default when creating our CustomKafkaListener instance:

public CustomKafkaListener(String topic, String bootstrapServers) {
    this(topic, defaultKafkaConsumer(bootstrapServers));
}

static KafkaConsumer<String, String> defaultKafkaConsumer(String boostrapServers) {
    Properties props = new Properties();
    props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, boostrapServers);
    props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test_group_id");
    props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    return new KafkaConsumer<>(props);
}

For this example, we hardcoded most of these properties, but, ideally, they should be loaded from a configuration file. Let’s quickly see what each of the properties means:

• Boostrap Servers: a list of host and port pairs used for establishing the initial connection to the Kafka cluster
• Group ID: an ID that allows a group of consumers to jointly consume a set of topic partitions
• Auto Offset Reset: the position in the Kafka log to start reading data when there is no initial offset
• Key/Value Deserializers: the deserializer classes for the keys and values. For our example, we’ll use String keys and values and the following deserializer: org.apache.kafka.common.serialization.StringDeserializer

With this minimal configuration, we’ll be able to subscribe to the topic and easily test out implementation. For a complete list of the available properties, consult the official documentation.

2.2. Subscribe to a Topic

Now, we need to provide a way to subscribe to the topic and start polling for messages. This can be achieved using KafkaConsumer‘s subscribe() method, followed by an infinite loop calling to the poll() method. Moreover, since this method will block the thread, we can implement the Runnable interface to provide a nice integration with CompletableFuture:

class CustomKafkaListener implements Runnable {
    private final String topic;
    private final KafkaConsumer<String, String> consumer;

    // constructors

    @Override
    void run() {
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            consumer.poll(Duration.ofMillis(100))
              .forEach(record -> log.info("received: " + record)); 
        } 
    }
}

Now, our CustomKafkaListener can be started  like this without blocking the main thread:

String topic = "baeldung.articles.published";
String bootstrapServers = "localhost:9092";

var listener = new CustomKafkaListener(topic, bootstrapServers)
CompletableFuture.runAsync(listener);

2.3. Consume Activity

Currently, our application only listens to the topic and logs all incoming messages. Let’s further improve it to allow more complex scenarios and make it more testable. For example, we can allow defining a Consumer<String> that will accept each new event from the topic:

class CustomKafkaListener implements Runnable {
    private final String topic;
    private final KafkaConsumer<String, String> consumer;
    private Consumer<String> recordConsumer;

    CustomKafkaListener(String topic, KafkaConsumer<String, String> consumer) {
        this.topic = topic;
        this.consumer = consumer;
        this.recordConsumer = record -> log.info("received: " + record);
    }

   // ...

    @Override
    public void run() {
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            consumer.poll(Duration.ofMillis(100))
              .forEach(record -> recordConsumer.accept(record.value()));
        }
    }

    CustomKafkaListener onEach(Consumer newConsumer) {
        recordConsumer = recordConsumer.andThen(newConsumer);
        return this;
    }
}

Declaring the recordConsumer as a Consumer<String> allows us to chain multiple functions using the default method andThen(). These functions will be called, one by one, for each incoming message.

3. Testing

To test our implementation, we’ll create a KafkaProducer and use it to publish some messages to the “baeldung.articles.published” topic. Then, we’ll start our CustomKafkaListener and verify it accurately processes all the activity.

3.1. Setup Kafka Testcontainer

We can utilize the Testcontainers library to spin up a Kafka container within our test environment. Firstly, we’ll need to add Testcontainer dependencies for the JUnit5 extension and the Kafka module:

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>kafka</artifactId>
    <version>1.19.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>1.19.3</version>
    <scope>test</scope>
</dependency>

We can now create a KafkaContainer with a specific Docker image name. Then, we’ll add the @Container and @Testcontainers annotations to allow the Testcontainers JUnit5 extension to manage the container’s lifecycle:

@Testcontainers
class CustomKafkaListenerLiveTest {

    @Container
    private static final KafkaContainer KAFKA_CONTAINER = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"));

    // ...
}

3.2. Create and Start the Listener

Firstly, we’ll define the topic name as a hardcoded String and extract the bootstrapServers from the KAFKA_CONTAINER. Additionally, we’ll create an ArrayList<String> that will be used for collecting the messages:

String topic = "baeldung.articles.published";
String bootstrapServers = KAFKA_CONTAINER.getBootstrapServers();
List<String> consumedMessages = new ArrayList<>();

We’ll create an instance of a CustomKafkaListener using these properties and instruct it to capture new messages and add them to the consumedMessages list:

CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers).onEach(consumedMessages::add);
listener.run();

However, it’s crucial to note that running it as is might block the thread and freeze the test. To prevent this, we’ll execute it asynchronously using a CompletableFuture:

CompletableFuture.runAsync(listener);

While not critical for testing, we can also instantiate the listener within a try-with-resources block in the first place:

var listener = new CustomKafkaListener(topic, bootstrapServers).onEach(consumedMessages::add);
CompletableFuture.runAsync(listener);

3.3. Publish Messages

To send article names to the “baeldung.articles.published” topic, we’ll set up a KafkaProducer using a Properties object, following a similar approach to what we did for the consumer.

static KafkaProducer<String, String> testKafkaProducer() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers());
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    return new KafkaProducer<>(props);
}

This method will allow us to publish messages to test our implementation. Let’s create another test helper that will send a message for each article received as a parameter:

private void publishArticles(String topic, String... articles) {
    try (KafkaProducer<String, String> producer = testKafkaProducer()) {
        Arrays.stream(articles)
          .map(article -> new ProducerRecord<String,String>(topic, article))
          .forEach(producer::send);
    }
}

3.4. Verify

Let’s put all the pieces together and run our test. We’ve already discussed how to create a CustomKafkaListener and start publishing data:

@Test
void givenANewCustomKafkaListener_thenConsumesAllMessages() {
    // given
    String topic = "baeldung.articles.published";
    String bootstrapServers = KAFKA_CONTAINER.getBootstrapServers();
    List<String> consumedMessages = new ArrayList<>();

    // when
    CustomKafkaListener listener = new CustomKafkaListener(topic, bootstrapServers).onEach(consumedMessages::add);
    CompletableFuture.runAsync(listener);
    
    // and
    publishArticles(topic,
      "Introduction to Kafka",
      "Kotlin for Java Developers",
      "Reactive Spring Boot",
      "Deploying Spring Boot Applications",
      "Spring Security"
    );

    // then
    // ...
}

Our final task involves waiting for the asynchronous code to finish and confirming that the consumedMessages list contains the expected content. To achieve this, we’ll employ the Awaitility library, utilizing its await().untilAsserted():

// then
await().untilAsserted(() -> 
  assertThat(consumedMessages).containsExactlyInAnyOrder(
    "Introduction to Kafka",
    "Kotlin for Java Developers",
    "Reactive Spring Boot",
    "Deploying Spring Boot Applications",
    "Spring Security"
  ));

4. Conclusion

In this tutorial, we learned how to use Kafka’s Consumer and Producer API without relying on higher-level Spring modules. First, we created a consumer using CustomKafkaListener that encapsulates a KafkaConsumer. For testing, we implemented a KafkaProducer and verified our setup using Testcontainers and Awaitility.