Careeroza

Kafka Core Concepts

Topics, Partitions, Offsets

ConceptMeaning
BrokerA Kafka server. A cluster has several brokers.
TopicA named stream of events, e.g. order-placed.
PartitionA topic is split into partitions (an append-only log each). More partitions = more parallel consumers.
OffsetPosition of a record inside a partition: 0, 1, 2... Records are kept after reading (retention, e.g. 7 days).
KeyRecords with the same key go to the same partition, so they stay in order (e.g. key = userId or orderId).
Consumer groupConsumers sharing a group.id split the partitions between them. Each group gets every record.
Committed offsetWhere a group has read up to, per partition. A restarted consumer continues from there.
ReplicationEach partition is copied to N brokers so a broker crash loses nothing.

An In-memory Topic

This plain-Java model of a topic shows keys choosing partitions, offsets, two consumer groups each receiving every event, and partitions split between the consumers of one group:

Example

import java.util.*;

public class Main {
    public static void main(String[] args) {
        Topic topic = new Topic("order-placed", 3);
        String[][] orders = {{"user-42", "1001", "2998"}, {"user-7", "1002", "1499"}, {"user-42", "1003", "799"},
                             {"user-13", "1004", "54999"}, {"user-7", "1005", "349"}, {"user-42", "1006", "1299"}};
        for (String[] o : orders) {
            OrderPlaced e = new OrderPlaced(Long.parseLong(o[1]), o[0], Integer.parseInt(o[2]));
            int[] pos = topic.send(o[0], e);
            System.out.println("producer: " + e + " -> partition " + pos[0] + ", offset " + pos[1]);
        }

        ConsumerGroup payments = new ConsumerGroup("payment-service", topic, List.of("payment-A", "payment-B"));
        ConsumerGroup notifications = new ConsumerGroup("notification-service", topic, List.of("notify-A"));
        payments.poll();
        notifications.poll();

        System.out.println("--- payment-service polls again: nothing new ---");
        payments.poll();
        topic.send("user-7", new OrderPlaced(1007, "user-7", 4999));
        System.out.println("--- one new event ---");
        payments.poll();
    }
}

record OrderPlaced(long orderId, String userId, int amount) {}
record Message(String key, OrderPlaced value) {}

class Topic {
    final List<List<Message>> partitions = new ArrayList<>();

    Topic(String name, int count) {
        for (int i = 0; i < count; i++) partitions.add(new ArrayList<>());
    }

    int[] send(String key, OrderPlaced value) {
        int p = Math.floorMod(key.hashCode(), partitions.size());   // same key -> same partition
        partitions.get(p).add(new Message(key, value));
        return new int[] {p, partitions.get(p).size() - 1};
    }
}

class ConsumerGroup {
    private final String groupId;
    private final Topic topic;
    private final Map<String, List<Integer>> assignment = new LinkedHashMap<>();
    private final int[] committed;

    ConsumerGroup(String groupId, Topic topic, List<String> consumers) {
        this.groupId = groupId;
        this.topic = topic;
        this.committed = new int[topic.partitions.size()];
        consumers.forEach(c -> assignment.put(c, new ArrayList<>()));
        for (int p = 0; p < committed.length; p++) assignment.get(consumers.get(p % consumers.size())).add(p);
        System.out.println(groupId + " assignment: " + assignment);
    }

    void poll() {
        assignment.forEach((consumer, parts) -> {
            for (int p : parts) {
                List<Message> log = topic.partitions.get(p);
                while (committed[p] < log.size()) {
                    System.out.println("  [" + groupId + "/" + consumer + "] p" + p + "@" + committed[p] + " " + log.get(committed[p]).value());
                    committed[p]++;   // commit the offset
                }
            }
        });
    }
}

Output

producer: OrderPlaced[orderId=1001, userId=user-42, amount=2998] -> partition 2, offset 0
producer: OrderPlaced[orderId=1002, userId=user-7, amount=1499] -> partition 0, offset 0
producer: OrderPlaced[orderId=1003, userId=user-42, amount=799] -> partition 2, offset 1
producer: OrderPlaced[orderId=1004, userId=user-13, amount=54999] -> partition 0, offset 1
producer: OrderPlaced[orderId=1005, userId=user-7, amount=349] -> partition 0, offset 2
producer: OrderPlaced[orderId=1006, userId=user-42, amount=1299] -> partition 2, offset 2
payment-service assignment: {payment-A=[0, 2], payment-B=[1]}
notification-service assignment: {notify-A=[0, 1, 2]}
  [payment-service/payment-A] p0@0 OrderPlaced[orderId=1002, userId=user-7, amount=1499]
  [payment-service/payment-A] p0@1 OrderPlaced[orderId=1004, userId=user-13, amount=54999]
  [payment-service/payment-A] p0@2 OrderPlaced[orderId=1005, userId=user-7, amount=349]
  [payment-service/payment-A] p2@0 OrderPlaced[orderId=1001, userId=user-42, amount=2998]
  [payment-service/payment-A] p2@1 OrderPlaced[orderId=1003, userId=user-42, amount=799]
  [payment-service/payment-A] p2@2 OrderPlaced[orderId=1006, userId=user-42, amount=1299]
  [notification-service/notify-A] p0@0 OrderPlaced[orderId=1002, userId=user-7, amount=1499]
  [notification-service/notify-A] p0@1 OrderPlaced[orderId=1004, userId=user-13, amount=54999]
  [notification-service/notify-A] p0@2 OrderPlaced[orderId=1005, userId=user-7, amount=349]
  [notification-service/notify-A] p2@0 OrderPlaced[orderId=1001, userId=user-42, amount=2998]
  [notification-service/notify-A] p2@1 OrderPlaced[orderId=1003, userId=user-42, amount=799]
  [notification-service/notify-A] p2@2 OrderPlaced[orderId=1006, userId=user-42, amount=1299]
--- payment-service polls again: nothing new ---
--- one new event ---
  [payment-service/payment-A] p0@3 OrderPlaced[orderId=1007, userId=user-7, amount=4999]

Note: Real Kafka uses the murmur2 hash of the key bytes, not String.hashCode(), but the rule is the same: same key → same partition → strict order for that key. There is no order across partitions.

Run Kafka Locally

Example

Kafka 3.x runs without ZooKeeper (KRaft mode). One container is enough for learning:

# docker-compose.yml
services:
  kafka:
    image: apache/kafka:3.8.0
    container_name: kafka
    ports:
      - "9092:9092"

docker compose up -d
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic order-placed --partitions 3 --replication-factor 1
docker exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic order-placed

Example output

Created topic order-placed.
Topic: order-placed	TopicId: q1Xb7mR2T0uY8kZp3nW5aA	PartitionCount: 3	ReplicationFactor: 1	Configs:
	Topic: order-placed	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: order-placed	Partition: 1	Leader: 1	Replicas: 1	Isr: 1
	Topic: order-placed	Partition: 2	Leader: 1	Replicas: 1	Isr: 1