Kafka - Consumers

A Kafka Consumer reads records from one or more topic partitions. Consumers belong to a consumer group: Kafka distributes partitions across the group members for parallel processing while ensuring each partition is consumed by exactly one member at a time.

Share

Contents

Basic Consumer Example

A consumer is a loop: poll() fetches the next records from the partitions assigned to it, the application processes them, and the loop calls poll() again. With the default settings, offsets are committed automatically:

Java
import java.time.Duration;
import java.util.List;
import java.util.Properties;

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

public class SimpleConsumer {

    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-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // start from the beginning if the group has no offset yet

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("my-topic"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("partition=%d offset=%d key=%s value=%s%n",
                            record.partition(), record.offset(), record.key(), record.value());
                }
                // With auto-commit (the default), the next poll() commits the offsets
                // of these records, at most every 5 seconds
            }
        }
    }
}

The other examples in this article show only the part that changes, inside this same main method.

Consumer Groups and Partition Assignment

All consumers sharing the same group.id form a consumer group. Kafka assigns partitions such that each partition is consumed by exactly one consumer in the group.

  • 3 partitions, 1 consumer → consumer reads all 3 partitions.
  • 3 partitions, 3 consumers → each consumer reads 1 partition (full parallelism).
  • 3 partitions, 4 consumers → one consumer gets no partition and stays idle until another leaves. Extra consumers add no parallelism, only spare capacity.
  • Different group.id → independent consumption; each group reads all messages (broadcast).
Use multiple consumer groups to have the same events consumed independently by different services (e.g., one group for analytics, another for real-time notifications).

Offsets and Commit Strategies

Each consumer tracks its position (offset) per partition. On restart, it resumes from the last committed offset.

StrategyConfig / APITrade-off
Auto-commitenable.auto.commit=true, auto.commit.interval.ms=5000Simple. Safe as long as you finish processing records before the next poll(); after a crash, up to 5 seconds of records are processed again
Manual sync commitconsumer.commitSync()Waits until the broker confirms the commit and retries on recoverable errors; slower, because the consumer stops while it waits
Manual async commitconsumer.commitAsync()Non-blocking; a failed commit is not retried, so a crash right after it means some records are processed again
Commit specific offsetscommitSync(offsets)You choose exactly which offset to commit for each partition, for example after every record. More commits mean more round trips to the broker

Manual Offset Commit

With auto-commit turned off, the application commits once a batch has been processed. The usual pattern commits asynchronously inside the loop, for speed, and synchronously once on shutdown, so the last position is not lost. wakeup() is the safe way to stop a consumer from another thread: it makes the blocked poll() throw WakeupException.

Java
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));

// Stop cleanly when the JVM shuts down
Thread mainThread = Thread.currentThread();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    consumer.wakeup();
    try { mainThread.join(); } catch (InterruptedException ignored) { }
}));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
        for (ConsumerRecord<String, String> record : records) {
            processRecord(record);
        }
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) {
                log.warn("Commit failed for {}", offsets, exception);
            }
        });
    }
} catch (WakeupException e) {
    // Expected on shutdown
} finally {
    try {
        consumer.commitSync();   // commit the last processed position
    } finally {
        consumer.close();
    }
}
Committing after processing gives at-least-once delivery: no message is lost, but after a crash some are processed again. Make processing idempotent so a repeat does no harm. Never commit after a batch that failed: commitSync() commits everything the last poll() returned, including the records that were not processed.

Rebalancing

A rebalance occurs when a consumer joins or leaves the group, or when a consumer crashes (detected via session timeout). Partitions are then reassigned. How disruptive that is depends on the protocol: with the original "eager" protocol, every consumer gives up all its partitions and consumption stops until the new assignment is ready. Cooperative assignment, and the new consumer group protocol that became generally available in Kafka 4.0 (group.protocol=consumer), move only the partitions that change owner, while the others keep being read.

  • Commit offsets before rebalance using ConsumerRebalanceListener.onPartitionsRevoked().
  • Tune session.timeout.ms (how long before Kafka considers a consumer dead) and heartbeat.interval.ms. With group.protocol=consumer, these two are set on the broker for the whole group instead.
  • max.poll.interval.ms: if poll() is not called within this time, consumer is kicked out of the group.
Java
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Commit offsets before losing partition ownership
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Optionally seek to a specific offset
    }
});

Key Configuration Summary
ConfigDefaultPurpose
group.id-Consumer group identifier
auto.offset.resetlatestWhere a group starts reading when it has no committed offset yet: earliest (oldest retained record) or latest (only new records)
enable.auto.committrueCommit offsets automatically during poll()
auto.commit.interval.ms5000Shortest time between two automatic commits
max.poll.records500Most records one poll() returns
session.timeout.ms45000How long the group waits without hearing from a consumer before removing it and giving its partitions to others
heartbeat.interval.ms3000How often the consumer tells the broker managing its group (the group coordinator) that it is still alive
max.poll.interval.ms300000Longest allowed gap between poll() calls before the consumer is removed from the group
group.protocolclassicconsumer selects the new rebalance protocol (Kafka 4.0+ brokers)