Kafka Java Consumer

A Java consumer pulls messages from Kafka topics and hands them to your business logic. The consumer asks the broker for new data in a repeating loop, so the program controls its own reading speed. Think of a librarian who visits a shelf at regular intervals, collects every new book, and records the last book number in a notebook so the next visit starts at the right place.

The Consumer Loop

  subscribe to topic
        |
        v
  +-----------------+
  |  poll(timeout)  | <---------+
  +-----------------+           |
        |                       |
        v                       |
  process each record           |
        |                       |
        v                       |
  commit offsets  --------------+

The loop runs until the application shuts down. Every pass fetches a batch of records, processes them, and saves the position.

A Complete Consumer

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class OrderConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "billing-app");
        props.put("key.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");
        props.put("enable.auto.commit", "false");

        try (KafkaConsumer<String, String> consumer =
                 new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));
            while (true) {
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> r : records) {
                    System.out.println(r.key() + " -> " + r.value()
                        + " (partition " + r.partition()
                        + ", offset " + r.offset() + ")");
                }
                consumer.commitSync();
            }
        }
    }
}

Key Settings Explained

Setting	Meaning
group.id	Name of the consumer group that shares the work
auto.offset.reset	Where to start when no saved offset exists: earliest or latest
enable.auto.commit	Lets the client save offsets on a timer when true
max.poll.records	Upper limit of records returned by one poll

Setting auto commit to false gives full control. The program commits only after processing succeeds, which prevents message loss during crashes.

Commit Choices

commitSync

Blocks until the broker confirms the commit. It retries on temporary errors and offers the strongest safety.

commitAsync

Returns immediately and reports the result through a callback. It gives higher throughput, but a failed commit does not retry automatically.

A common pattern uses commitAsync inside the loop and one final commitSync during shutdown.

Shut Down Cleanly

Call wakeup from a shutdown hook to break a blocked poll call. The loop catches the WakeupException, commits offsets, and closes the consumer so the group rebalances right away.

Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));

try {
    while (true) { /* poll and process */ }
} catch (WakeupException e) {
    // expected during shutdown
} finally {
    consumer.commitSync();
    consumer.close();
}

Thread Safety

A KafkaConsumer is not thread-safe. Keep one consumer per thread. Scale reading by starting more consumer instances in the same group, up to the number of partitions.

Assign Partitions by Hand

The assign method skips group management and attaches the consumer to specific partitions. Use it for tools that must read exact partitions, such as replay utilities. Regular applications prefer subscribe.

Seek to a Position

consumer.assign(List.of(new TopicPartition("orders", 0)));
consumer.seek(new TopicPartition("orders", 0), 150L);

Seeking lets a program replay old data from a chosen offset, which helps with bug fixes and audits.

Avoid Slow Processing

Long work between poll calls can exceed max.poll.interval.ms and push the consumer out of the group. Reduce max.poll.records or move heavy tasks to a worker pool so the poll loop stays fast.