0Pricing
Apache Kafka & Stream Processing Fundamentals · Lesson

Consuming Messages from Kafka

Learn to build consumers that read and process data streams from Kafka topics, handling offsets and consumer groups.

Consuming Messages from Kafka is a free Apache Kafka & Stream Processing Fundamentals lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the Apache Kafka & Stream Processing Fundamentals learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

What are Kafka Consumers?

Kafka Consumers are applications that read data from Kafka topics. They subscribe to one or more topics and process the messages as they arrive.

  • Consumers are the 'listeners' in your Kafka ecosystem.
  • They pull data from Kafka brokers, rather than having data pushed to them.
  • Essential for building real-time data pipelines and analytics systems.

Working with Consumer Groups

Consumers often work together in a consumer group. This is a collection of consumer instances that share a common group.id.

  • Scalability: Messages from a topic's partitions are distributed among consumers in the same group, allowing parallel processing.
  • Fault Tolerance: If a consumer instance fails, its assigned partitions are automatically reassigned to other active consumers in the group.
  • Each message in a partition is delivered to only one consumer within a group.

How Consumers Read Data (Polling)

Kafka consumers don't just 'receive' messages. Instead, they actively poll Kafka brokers for new data.

  • The consumer repeatedly asks Kafka, "Do you have any new records for me?"
  • This polling mechanism gives the consumer control over its processing rate.
  • If no new records are available, the poll() method will simply return an empty list after waiting for a specified duration.

Basic Java Consumer Setup

To create a Kafka consumer in Java, you use the KafkaConsumer class. First, you need to set up some basic properties like the Kafka server address and deserializers.

Here's the start of a simple consumer program:

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;
import java.util.Collections;

public class SimpleConsumerSetup {
    public static void main(String[] args) {
        // 1. Define consumer properties
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-first-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 2. Create the consumer
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 3. Subscribe to a topic
        consumer.subscribe(Collections.singletonList("my-topic"));

        // Consumer loop and closing will go here
        System.out.println("Consumer setup complete. Subscribed to 'my-topic'.");
        consumer.close();
    }
}

Essential Consumer Properties

Let's break down the key properties we set for our consumer:

  • BOOTSTRAP_SERVERS_CONFIG: A list of Kafka broker addresses (e.g., localhost:9092) that the consumer will connect to.
  • GROUP_ID_CONFIG: A unique identifier for the consumer group this consumer belongs to (e.g., my-first-group).
  • KEY_DESERIALIZER_CLASS_CONFIG: Specifies how to convert message keys from bytes (received from Kafka) into Java objects (e.g., StringDeserializer for text keys).
  • VALUE_DESERIALIZER_CLASS_CONFIG: Specifies how to convert message values from bytes into Java objects (e.g., StringDeserializer for text values).

The Consumer Loop: Polling for Records

Consumers continuously poll Kafka for new data. This is typically done in an infinite loop. The poll(Duration timeout) method returns a batch of ConsumerRecords or an empty list if no new records are available within the specified timeout.

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Properties;
import java.util.Collections;

public class PollingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-polling-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        try {
            while (true) { // Infinite loop to continuously poll
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    System.out.println("No records received. Waiting...");
                } else {
                    System.out.println("Received " + records.count() + " records.");
                    // Processing of records would happen here
                }
            }
        } finally {
            consumer.close();
            System.out.println("Consumer closed.");
        }
    }
}

Processing Received Messages

Once you receive ConsumerRecords from poll(), you can iterate over them to process each message individually. Each ConsumerRecord contains the topic, partition, offset, key, and value of the message.

This is where your application's specific logic for handling the data lives.

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Properties;
import java.util.Collections;

public class ProcessingConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
                    // Add your custom processing logic here, e.g., store in database, perform analytics
                }
            }
        } finally {
            consumer.close();
            System.out.println("Consumer closed.");
        }
    }
}

Tracking Progress with Offsets

An offset is a unique, sequential ID assigned to each message within a Kafka topic partition. It acts like a pointer, indicating the position of a message.

  • Consumers use offsets to track which messages they have already processed.
  • Kafka stores the latest committed offset for each consumer group and partition.
  • This ensures that if a consumer restarts or a partition is reassigned, it knows exactly where to resume reading to avoid reprocessing or missing messages.

Offsets are crucial for reliable message processing.

Automatic Offset Committing

By default, Kafka consumers are configured to auto-commit offsets periodically. This is convenient for simple applications but might lead to message loss or duplicates in certain failure scenarios.

  • Set enable.auto.commit to true (this is the default setting).
  • The auto.commit.interval.ms property determines how often offsets are committed (e.g., every 5 seconds).
  • This approach is generally suitable for 'at-least-once' processing where some duplicates are acceptable.

Manual Offset Control

For more precise control over message processing and stronger guarantees (like 'exactly-once' processing), you can manually commit offsets after your application has successfully processed a batch of messages.

  • Set enable.auto.commit to false to disable automatic commits.
  • Use consumer.commitSync(): This method blocks until the commit operation is successful.
  • Use consumer.commitAsync(): This method is non-blocking and faster, but requires a callback function for handling potential errors.

Manual commits ensure that records are processed *before* their offsets are marked as 'done', preventing data loss on consumer failure.

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Properties;
import java.util.Collections;

public class ManualCommitConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-manual-commit-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Disable auto-commit

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Processed: Partition %d, Offset %d, Value: %s%n",
                        record.partition(), record.offset(), record.value());
                    // Simulate processing work that must complete before committing
                }
                if (!records.isEmpty()) {
                    consumer.commitSync(); // Commit offsets after processing the batch
                    System.out.println("Offsets committed manually.");
                }
            }
        } finally {
            consumer.close();
            System.out.println("Consumer closed.");
        }
    }
}

Check Your Understanding

Consider a Kafka topic with 3 partitions. A consumer group with 2 consumer instances is subscribed to this topic. If one consumer instance fails, what happens to the partitions it was processing, and how does Kafka ensure no data is lost?

Lesson Recap: Consuming Data

Great job! In this lesson, you learned how Kafka consumers read data:

  • Consumers subscribe to topics and actively poll Kafka for new messages.
  • Consumer groups enable scalable and fault-tolerant message processing across multiple consumer instances.
  • Offsets are sequential IDs that track a consumer's progress within a topic partition.
  • You can choose between automatic (simpler, default) or manual (more control, better guarantees) offset committing for reliability.

Next, we'll dive deeper into partitions and offsets to understand how Kafka achieves high throughput and message ordering.

Frequently asked questions

Is the “Consuming Messages from Kafka” lesson free?

Yes — the full text of “Consuming Messages from Kafka” is free to read here on the web, and the Apache Kafka & Stream Processing Fundamentals course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the Apache Kafka & Stream Processing Fundamentals course, upgrade to CoddyKit PRO.

What will I learn in “Consuming Messages from Kafka”?

Learn to build consumers that read and process data streams from Kafka topics, handling offsets and consumer groups. You practise Apache Kafka & Stream Processing Fundamentals with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start Apache Kafka & Stream Processing Fundamentals?

No prior experience is required. Apache Kafka & Stream Processing Fundamentals on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Consuming Messages from Kafka” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this Apache Kafka & Stream Processing Fundamentals lesson?

Yes. Every Apache Kafka & Stream Processing Fundamentals lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Producing Messages to Kafka
  2. Consuming Messages from Kafka
  3. Understanding Partitions & Offsets
  4. Message Keys and Partitioning Strategies
← Back to Apache Kafka & Stream Processing Fundamentals