01 - Apache Kafka Fundamentals: Topics, Partitions, Offsets and Consumer Groups

Why a queue that deletes is not enough

Picture an online shop. Every order must reach the warehouse, invoicing and the service that computes statistics, and each one works at its own pace: the warehouse in real time, the statistics once an hour. In three months a fourth service will arrive, for purchase suggestions, and it will want to reread past orders too. With a traditional queue, where the message disappears once a consumer acknowledges it, the fourth service starts from zero and the past is lost.

Apache Kafka solves the problem by changing the model: not a queue that empties, but a log where records are always appended at the end and stay in place until the retention expires. Reading deletes nothing. Each reader remembers on its own how far it has got and can go back. This is the difference that holds up the rest of the article, and the introduction of the official documentation presents it the same way: events kept for as long as needed and reread when required.

A note, so as not to wrong classic queues: RabbitMQ too now offers streams, logs that can be reread until they expire, so the right comparison is not "Kafka versus queues" but "log versus queue with acknowledgement". In this first article of the series you build the foundations: topics, partitions, offsets, consumer groups, replication, retention. For each one you find a small Java 21 model, which I compiled and ran, and the configuration values read in the Kafka 4.4 documentation (the current version, released on 22 October 2026, consulted on 2026-11-12).

Topics and partitions: the log that is only written at the end

A topic is the name under which producers publish and consumers read, for example orders-placed. It is split into partitions, and the partition is the unit that really matters: an ordered, immutable log in which every record receives a progressive number called an offset (0, 1, 2, ...). Order is guaranteed inside the partition and not across different partitions: the documentation says that "events with the same event key (e.g., a customer or vehicle ID) are written to the same partition" and that whoever reads a partition sees the events in the order in which they were written.

How does the producer choose the partition? If the record has a key, the choice depends on a hash of the key; if the key is missing, the default partitioner fills one partition up to batch.size bytes and then moves to another (this is the description of partitioner.class in the producer documentation). The model below reproduces the first rule with the string hashCode and the remainder of the division by the number of partitions. Kafka uses a different hash function, so the partitions chosen by a real broker will differ, but the property that matters is the same: same key, same partition.

import java.util.*;

public class Partitions {
    record Record(long offset, String key, String value) {}

    static class Topic {
        final List<List<Record>> partitions = new ArrayList<>();
        Topic(int n) { for (int i = 0; i < n; i++) partitions.add(new ArrayList<>()); }

        static int where(String key, int n) {            // model: hash of the key modulo n
            return Math.floorMod(key.hashCode(), n);
        }
        void write(String key, String value) {
            List<Record> log = partitions.get(where(key, partitions.size()));
            log.add(new Record(log.size(), key, value)); // the offset is the position in the log
        }
    }

    public static void main(String[] args) {
        Topic orders = new Topic(3);
        String[] customers = {"anna", "marco", "anna", "luca", "marco", "anna", "luca", "anna"};
        for (int i = 0; i < customers.length; i++) orders.write(customers[i], "order-" + i);

        for (int p = 0; p < 3; p++) {
            System.out.print("partition " + p + ":");
            for (Record r : orders.partitions.get(p))
                System.out.print(" [" + r.offset() + " " + r.key() + " " + r.value() + "]");
            System.out.println();
        }

        int moved = 0, total = 1000;
        for (int k = 0; k < total; k++) {
            String key = "customer-" + k;
            if (Topic.where(key, 6) != Topic.where(key, 12)) moved++;
        }
        System.out.println("keys that change partition going from 6 to 12: " + moved + " out of " + total);
    }
}

// Output:
// partition 0: [0 anna order-0] [1 anna order-2] [2 anna order-5] [3 anna order-7]
// partition 1: [0 luca order-3] [1 luca order-6]
// partition 2: [0 marco order-1] [1 marco order-4]
// keys that change partition going from 6 to 12: 502 out of 1000

The where method is the whole of partitioning: with three partitions, anna always ends up in 0, luca in 1 and marco in 2. Look at the offsets: in partition 0 Anna's four orders have offsets 0, 1, 2, 3 and appear in the order in which they were sent (order-0, order-2, order-5, order-7). This is what "ordering by key" means: her orders will never swap places, while between Anna and Marco no order is guaranteed. The last line of the output is the trap: going from 6 to 12 partitions, 502 keys out of 1000 change partition (about half, as expected, since the remainder modulo 12 equals the remainder modulo 6 only half of the time). The output is deterministic because String.hashCode is specified.

From here follows a practical rule that no cluster reminds you of. The operations documentation describes how to increase partitions (kafka-topics.sh --alter --partitions) and warns that, if the data is partitioned with hash(key) modulo the number of partitions, the mapping changes: records with the same key may end up in different partitions after the expansion, with an effect on ordering for the keys that already exist, and Kafka does not redistribute the data already written. That is why it pays to choose the number with some margin from the start, reasoning on two quantities you can measure: how many consumers you want to run in parallel (you will not need more than the partitions, see below) and how many MB per second the topic must sustain, divided by how much one partition sustains on your hardware. No "typical" number replaces a test on your disk and your network.

Broker, replication and the moment a record is "written"

A broker is a Kafka server. Every partition has a leader, which serves reads and writes, and zero or more followers that copy the leader's log. The set of replicas that keep up is called the ISR (in-sync replicas). A follower leaves the ISR if it does not fetch data or does not reach the end of the leader's log for longer than replica.lag.time.max.ms, which by default is 30 seconds (broker documentation, 4.4).

The definition that supports the rest comes from the "Design" section of the documentation: a record is committed when all the replicas of the ISR have applied it, and a committed record is not lost as long as at least one of those replicas stays alive. Consumers do not see a record before that moment, nor if the in-sync replicas are fewer than min.insync.replicas. The default of min.insync.replicas is 1: if you want acks=all to really mean "at least two copies", with replication factor 3 set it to 2 at topic level. The cost is availability: with two brokers down out of three the topic rejects writes, but loses nothing.

The offsets you will meet in commands and charts have similar names and different meanings. It pays to separate them right away:

• Log end offset (LEO): the offset that the next record written to a replica's log will have.
• High watermark (HW): the visibility limit. By convention it is the offset of the first record not yet visible to consumers, that is, everything before it is already replicated on the ISR.
• Current offset: the next record that this consumer will read with the following poll().
• Committed offset: the value saved for the group in the internal topic __consumer_offsets; it is the point from which you restart after a crash or a change of assignment.

The first two live on the broker and say how far the log has advanced and how much of it is visible; the other two live on the consumer side and say how much has been read and how much has been saved. The lag you will see in dashboards is the distance between the LEO and the group's committed offset: the bigger it is, the further behind the group. Beware of a common mistake: the committed offset is not "the last record processed" but the next one to read, that is, last processed plus one, as the KafkaConsumer Javadoc specifies: "the committed offset should be the next message your application will consume".

import java.util.*;

public class OffsetCommit {
    public static void main(String[] args) {
        // The leader has written 10 records (offsets 0..9), the LEO is 10.
        // The ISR followers have copied up to 8 and 9 records: the slowest one sets the high watermark.
        long leo = 10;
        long[] followerLeo = {8, 9};
        long hw = Math.min(leo, Math.min(followerLeo[0], followerLeo[1]));
        System.out.println("LEO=" + leo + " HW=" + hw + " -> the consumer sees offsets 0.." + (hw - 1));

        // The log is now fully replicated: 10 visible records.
        long committed = 0;                      // saved offset: the NEXT record to read
        long lag = leo - committed;
        System.out.println("committed=" + committed + " lag=" + lag);

        // Scene 1: processes records 0..5, then COMMIT, then stops: restart without duplicates.
        for (long o = 0; o <= 5; o++) System.out.println("  processing offset " + o);
        committed = 5 + 1;                       // last processed + 1
        System.out.println("commit -> " + committed + ", lag=" + (leo - committed));

        // Scene 2: processes 6..8, then the process dies BEFORE the commit.
        for (long o = 6; o <= 8; o++) System.out.println("  processing offset " + o);
        System.out.println("crash before the commit: the saved offset stays " + committed);

        // On restart the consumer reads from the saved offset: 6, 7, 8 are processed again.
        System.out.println("restart from offset " + committed + ": 6, 7, 8 arrive again");
    }
}

// Output:
// LEO=10 HW=8 -> the consumer sees offsets 0..7
// committed=0 lag=10
//   processing offset 0
//   processing offset 1
//   processing offset 2
//   processing offset 3
//   processing offset 4
//   processing offset 5
// commit -> 6, lag=4
//   processing offset 6
//   processing offset 7
//   processing offset 8
// crash before the commit: the saved offset stays 6
// restart from offset 6: 6, 7, 8 arrive again

In the first part the high watermark is 8, the minimum among the replicas' LEOs (10, 9 and 8): the consumer sees offsets 0 to 7 even though the leader has written 10. In the second part the consumer processes up to offset 5 and saves 5 + 1: that "plus one" is the line that counts. Then it processes 6, 7 and 8 and dies before the commit. On restart it reads from the saved offset and those three records arrive again. This is the at-least-once semantics of the documentation: no loss, possible duplicates. A real broker does in addition the commit to __consumer_offsets and the automatic reassignment of the partition to another consumer of the group; the logic of the saved offset, however, is exactly this.

The consumer group: how reading scales

All consumers with the same group.id form a group and divide the topic's partitions among themselves. There is a single rule: a partition is assigned to only one consumer of the group at a time. It follows that the maximum parallelism of a group is the number of partitions: with six partitions, the seventh consumer does no work, waits, and takes over if another one fails. Different groups are independent: each has its own offsets and receives all the records of the topic, which is how warehouse and invoicing read the same stream without disturbing each other. What a domain event is and why an order publishes a fact instead of calling other services is in the article on the fundamentals of event-driven architecture; why the same record can reach a group twice and how to defend against it is in the article on consumer idempotency.

import java.util.*;

public class ConsumerGroup {
    // Model of the "range" assignment (same idea as Kafka's RangeAssignor):
    // each consumer receives n/m consecutive partitions, the first n%m receive one more.
    static Map<String, List<Integer>> assign(int partitions, List<String> members) {
        Map<String, List<Integer>> map = new LinkedHashMap<>();
        int base = partitions / members.size(), extra = partitions % members.size(), p = 0;
        for (int i = 0; i < members.size(); i++) {
            List<Integer> mine = new ArrayList<>();
            for (int k = 0; k < base + (i < extra ? 1 : 0); k++) mine.add(p++);
            map.put(members.get(i), mine);
        }
        return map;
    }

    public static void main(String[] args) {
        for (int m : new int[] {1, 3, 4, 9}) {
            List<String> members = new ArrayList<>();
            for (int i = 1; i <= m; i++) members.add("c" + i);
            System.out.println(m + " consumers on 6 partitions -> " + assign(6, members));
        }
        // Two groups on the same topic: each has its own assignments and its own offsets.
        System.out.println("inventory group -> " + assign(6, List.of("i1", "i2")));
        System.out.println("billing group   -> " + assign(6, List.of("f1", "f2", "f3")));
    }
}

// Output:
// 1 consumers on 6 partitions -> {c1=[0, 1, 2, 3, 4, 5]}
// 3 consumers on 6 partitions -> {c1=[0, 1], c2=[2, 3], c3=[4, 5]}
// 4 consumers on 6 partitions -> {c1=[0, 1], c2=[2, 3], c3=[4], c4=[5]}
// 9 consumers on 6 partitions -> {c1=[0], c2=[1], c3=[2], c4=[3], c5=[4], c6=[5], c7=[], c8=[], c9=[]}
// inventory group -> {i1=[0, 1, 2], i2=[3, 4, 5]}
// billing group   -> {f1=[0, 1], f2=[2, 3], f3=[4, 5]}

With one consumer all six partitions are its own; with three, two each; with four the division is not exact and the first two take two, the others one; with nine, three consumers are left with an empty list. It is the proof that adding consumers beyond the number of partitions achieves nothing. The last pair of lines shows two groups on the same topic with independent assignments. The model copies the logic of the RangeAssignor; remember, though, that in the classic protocol the default list of partition.assignment.strategy is [RangeAssignor, CooperativeStickyAssignor] and the first one applies, while with group.protocol=consumer (the new protocol, KIP-848) the assignment is decided by the coordinator on the broker. What happens when a consumer joins or leaves, that is, the rebalance, is the subject of the article on advanced producers and consumers.

Retention and compaction: how much stays in the log

Records are not deleted on consumption, so something must decide when they disappear. The topic settings ("Topic Configs" documentation, 4.4) are two policies:

• cleanup.policy=delete with retention.ms (default 604800000, that is 7 days) and retention.bytes (default -1, no limit): the oldest segment disappears when it exceeds the time or the size.
• cleanup.policy=compact: for each key only the last record remains, with delete.retention.ms (default 1 day) setting how long tombstones survive, that is, records with a null value that delete a key.

The two policies answer different questions. Time-based retention says "keep the history for seven days" and suits event streams such as clicks or orders; compaction says "keep the current state of every key" and suits master data, prices and configurations. Note that deletion acts per segment (default 1 GiB or 7 days, segment.bytes and segment.ms) and never on the active segment: a topic with little data will see nothing disappear until the segment closes.

import java.util.*;

public class Compaction {
    record Record(long offset, String key, String value) {}

    // Model of compaction: keeps the last record for each key, without changing the offsets.
    static List<Record> compact(List<Record> log) {
        Map<String, Long> last = new HashMap<>();
        for (Record r : log) last.put(r.key(), r.offset());
        List<Record> clean = new ArrayList<>();
        for (Record r : log) if (last.get(r.key()) == r.offset()) clean.add(r);
        return clean;
    }

    public static void main(String[] args) {
        List<Record> log = new ArrayList<>();
        String[][] writes = {
            {"user123", "mario@example.com"}, {"user456", "anna@example.com"},
            {"user123", "mario.rossi@example.com"}, {"user456", null}};   // null = tombstone
        for (String[] s : writes) log.add(new Record(log.size(), s[0], s[1]));

        System.out.println("before: " + log);
        System.out.println("after:  " + compact(log));
    }
}

// Output:
// before: [Record[offset=0, key=user123, value=mario@example.com], Record[offset=1, key=user456, value=anna@example.com], Record[offset=2, key=user123, value=mario.rossi@example.com], Record[offset=3, key=user456, value=null]]
// after:  [Record[offset=2, key=user123, value=mario.rossi@example.com], Record[offset=3, key=user456, value=null]]

The initial log has four records on two keys. After compaction offset 2 remains for user123 (the updated value) and offset 3 for user456, which is a tombstone: the value is null, but the record stays in the log for delete.retention.ms, so late consumers have time to see the deletion. The model also keeps the original offsets (2 and 3, not 0 and 1), as the documentation guarantees: "the offset for a message never changes". A real broker compacts in the background with the log cleaner threads and never touches the active segment; your consumer that starts from zero rebuilds the state by reading one record per key.

Internal topics and three commands to look inside

Kafka keeps its own state in topics named like yours. __consumer_offsets holds the committed offsets (50 partitions by default, offsets.topic.num.partitions; the group is assigned to one of them based on its group.id); __transaction_state holds the state of transactions (50 partitions, transaction.state.log.num.partitions); in KRaft mode the cluster metadata lives in the __cluster_metadata log (the documentation mentions it in the kafka-dump-log.sh commands). Both parameters set to 50 carry the note "should not change after deployment": they are choices to make before the first start.

The commands that follow are the template for creating a topic with six partitions and three replicas, describing it to see the leader, replicas and ISR of each partition, and measuring a group's lag. I did not run them, because there is no cluster in this environment, and I do not report an invented output: the output format changes with the version, and the column to look at first is LAG.

# Template to adapt: needs a cluster reachable at kafka1:9092
kafka-topics.sh --create \
  --bootstrap-server kafka1:9092 \
  --topic orders-placed \
  --partitions 6 \
  --replication-factor 3 \
  --config min.insync.replicas=2

kafka-topics.sh --describe \
  --bootstrap-server kafka1:9092 \
  --topic orders-placed

kafka-consumer-groups.sh \
  --bootstrap-server kafka1:9092 \
  --describe \
  --group inventory-service

The topic --describe is the first diagnosis when something goes wrong: if for a partition the ISR is shorter than the replicas, a follower has fallen behind. The group --describe shows, for each partition, the current offset, the log end offset and the difference. If the lag grows at every check the group cannot keep up: add consumers (up to the number of partitions) or make processing faster. If the group shows up as empty, it has no active members.

What you can do now

You can explain why Kafka is not a queue that empties and why the partition is the unit of ordering and of parallelism. You can predict where a record with a given key ends up and why changing the number of partitions moves the keys. You distinguish LEO, high watermark, current offset and committed offset, and you know that the last one is the next to read. You know how many consumers make sense for a group and that two groups on the same topic do not disturb each other. The next step is to understand who coordinates the cluster now that ZooKeeper is gone, in the article on KRaft, and how writing and reading are made reliable, in the article on transactions. If you arrive from another point of the series, the links are: Kafka Streams, Kafka Connect, monitoring, Kafka in production.