How to Consume Messages from Kafka

Apache Kafka is a powerful distributed streaming platform that allows you to publish and consume streams of data efficiently. In this tutorial, we will guide you through consuming messages from Kafka topics with a focus on understanding key concepts and performing each step seamlessly.

Prerequisites

• Apache Kafka installed and running. You can refer to our installation guide.
• Basic understanding of Kafka and its components.
• Java installed on your system as Kafka typically requires Java.

Step-by-Step Instructions

1. Understand Kafka Consumer Group

Kafka consumers read data from Kafka topics. When you have multiple consumers, they typically form a consumer group, which allows each message to be processed by only one consumer in the group.

2. Set up Your Environment

Ensure Kafka and ZooKeeper services are running:

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

3. Create a Kafka Consumer

Create a simple Java application that implements KafkaConsumer to read messages. Here’s a basic structure:

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Properties;
import java.util.Arrays;

public class KafkaMessageConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer consumer = new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList("your-topic"));

        while (true) {
            ConsumerRecords records = consumer.poll(100);
            for (ConsumerRecord record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}

4. Run Your Consumer

Compile and run your Java application:

javac -cp path-to-kafka-jars/*:. KafkaMessageConsumer.java
java -cp path-to-kafka-jars/*:. KafkaMessageConsumer

Your consumer will start logging messages it reads from the specified Kafka topic.

Troubleshooting Common Issues

• Consumer Not Reading Messages: Check if the topic exists and is populated with messages. Ensure your consumer is correctly configured and subscribed to the correct topic.
• Serialization Error: Ensure the serializers for both key and value are set correctly.
• Connection Issues: Verify Kafka and ZooKeeper services are running, and network configurations are correct.

Summary Checklist

• Ensure Kafka services are running.
• Create a Kafka Consumer in your preferred programming language.
• Configure consumer properties correctly.
• Subscribe to the designated Kafka topic.
• Poll for and process messages continuously.

With these steps, you should be able to effectively consume messages from Kafka and incorporate them into your data processing workflows.

About the Author

Dorian Kane

Author

View All Posts