Consumers and Consumer Groups

Key points

• A consumer group is a set of consumers sharing a group.id. Kafka gives each partition to exactly one member of the group.
• More consumers than partitions? The extras sit idle. Scaling out past the partition count buys you nothing.
• Two different groups reading one topic each receive every record. That is how you fan out to several services.
• Progress is a committed offset per partition, stored in the internal __consumer_offsets topic.
• enable.auto.commit=true commits in the background every 5s. Easy, but it can commit records you have not finished processing.
• Commit after the work, not before. That is the whole difference between losing records and reprocessing them.

Example

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class OrderConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing-service");
        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");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        try (Consumer<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) {
                    handle(r.key(), r.value());
                }
                consumer.commitSync();   // only after the whole batch succeeded
            }
        }
    }

    static void handle(String key, String value) { /* your business logic */ }
}

One partition, one consumer per group. Everything else follows from that rule.