Introduction to Kafka

Apache Kafka is a widely used implementation of the log-based queue from the Log-Based Message Queues lesson. The concepts from that lesson carry over directly: topics split into partitions, a key hash that picks the partition, consumer groups with one committed offset per partition, retention, and replay. Kafka adds replication across machines, producer durability settings, log compaction, and surrounding components such as KRaft, Kafka Connect, and Schema Registry.

Kafka's Names for the Log

A Kafka record contains a key, a value, a timestamp, and optional headers (Kafka introduction). The value holds the payload, such as a JSON-encoded BookingCreated event, while the key determines the target partition. Headers carry metadata like a trace ID without modifying the value. Because Kafka stores the key and value as raw bytes without reading them, the producer and consumer must agree on the data format.

Brokers and Partition Leaders

A Kafka cluster runs across several brokers. Each partition has multiple copies, called replicas, stored on different brokers. The number of copies is the topic's replication factor, and the Kafka docs describe 3 as a common production setting (introduction). One replica per partition serves as the leader, while the others act as followers. Producers write to the leader, and consumers usually read from the leader. Followers replicate the leader's log: they "consume messages from the leader just as a normal Kafka consumer would" (Kafka design).

Spreading partition leaders across brokers distributes the write load across the cluster. The diagram shows a bookings topic with 3 partitions and a replication factor of 3 across three brokers, read by an email consumer group with three consumers.

The dashed arrows show P0's followers copying from its leader; P1 and P2 replicate the same way. If broker 1 fails, P0 needs a new leader on broker 2 or broker 3. The following sections explain which replicas can take over and what happens to records that broker 1 had already accepted.

How the Producer Picks a Partition

When a record has a key, the default producer chooses "a partition based on a hash of the key" (producer configs). Every record with the key booking-21 lands on the same partition, keeping events for booking 21 in order. The Kafka Exercise lesson computes this hash for real keys. Changing the partition count alters the hash mapping, so adding partitions to a topic can route new records for a key to a different partition than its older records.

A keyless record is not tied to a specific partition. The producer groups records bound for the same partition into a batch and sends each batch in one request, with batch.size capping the batch in bytes (producer configs). How the producer routes keyless records determines how full each batch gets.

Older producers distributed keyless records round robin, creating many small batches and extra requests. In Kafka 2.4, KIP-480 changed the default to a sticky partitioner (Confluent blog). The producer sends keyless records to one partition until that batch is sent, then switches to another partition. KIP-794 refined this behavior. The current producer switches partitions after sending batch.size bytes to a partition, and it routes fewer records to partitions on slower brokers. Over time, keyless records spread across all partitions without ordering between them. Setting partitioner.ignore.keys=true tells the producer to ignore keys and apply this batching logic to keyed records as well.

Replication and the In-Sync Replica Set

Each leader tracks which followers are caught up. Together, the leader and these up-to-date followers form the in-sync replica set (ISR). A follower drops out of the ISR if it fails to fetch or catch up to the leader's log end within replica.lag.time.max.ms, which defaults to 30 seconds (broker configs). Once the follower catches up, it rejoins the ISR.

A record becomes committed when every replica in the ISR has written it (design). The high watermark is the offset up to which every ISR replica has copied the log. Consumers receive only committed records, reading up to the high watermark. When a leader fails, the controller (the component that manages cluster metadata, covered in the KRaft section) elects a new leader from the ISR. As the docs note, "only members of this set are eligible for election as leader." Because every ISR member holds every committed record, the new leader already has them. With a replication factor of f + 1, a topic can lose f replicas without losing committed records.

A replica outside the ISR can become leader only through unclean leader election, controlled by unclean.leader.election.enable (false by default) (topic configs). Under the default setting, if every ISR replica goes down, the partition has no leader and rejects writes until an ISR replica returns. Enabling unclean leader election keeps the partition available, though the new leader can lack records that the previous leader had committed.

The diagram shows one write with acks=all, a producer setting from the next section that makes the leader reply only after every ISR replica has the record.

Producer acks and min.insync.replicas

The producer acks setting controls when the leader replies that a write succeeded (producer configs):

acks=all still leaves one gap. The ISR can shrink to just the leader, for example when both followers fall behind or go down. At that point, "every ISR replica" means a single copy, so a crash on that broker loses the record or takes the partition offline. The topic setting min.insync.replicas closes this gap. With acks=all, the leader rejects a write when the ISR has fewer replicas than this minimum and returns a NotEnoughReplicas error (topic configs). The partition refuses writes rather than accepting a write it could lose (design).

The defaults are acks=all on the producer and min.insync.replicas=1 on the topic. For important data, a common configuration pairs a replication factor of 3 with min.insync.replicas=2. Writes succeed while any 2 of the 3 replicas remain in sync, so the topic survives one broker failure without losing acknowledged writes or halting writes. The trade-off with acks=all is latency, since the leader's reply waits for the slowest in-sync follower.

The widget below runs one partition with a replication factor of 3. Broker 2 copies from the leader 1 step behind, and broker 3 copies 2 steps behind. Start with acks = 1, press Play, and after a few records press "Crash the leader". Then press "Restart stopped brokers". Next, press Reset, choose acks = all, and crash the leader again. Last, press Reset with acks = all and min.insync.replicas = 2. Press "Stop a follower", wait until broker 3 leaves the ISR, and then press it again. Then press "Restart stopped brokers".

With acks = 1, crashing the leader loses the newest record because the old leader acknowledged the write before broker 2 (the new leader) could copy it. When the old leader restarts, it deletes that uncopied record so its log matches the new leader. With acks = all, the producer gets an acknowledgment only after broker 3 also copies the record, ensuring the new leader holds every acknowledged write. The producer retries in-flight records that lack an acknowledgment. Enabled by default, the idempotent producer attaches sequence numbers so retried records are not written twice (design). Stopping one follower pauses acks = all writes until that follower drops out of the ISR. Stopping the second follower shrinks the ISR to 1, falling below min.insync.replicas = 2 and causing the leader to reject every write. When the stopped brokers restart, they catch up with the leader, rejoin the ISR, and writes resume.

Consumer Groups, Committed Offsets, and Rebalancing

Consumers with the same group.id form a consumer group where "each partition is assigned to exactly one consumer in the group" (KafkaConsumer docs), following the rule from the previous lesson. A consumer commits its offsets to Kafka either automatically in the background (the default, every 5 seconds) or by calling a commit method after processing records (consumer configs). One broker acts as the group coordinator for each group. Confluent's documentation notes that Kafka hashes the group ID to pick a partition in the internal __consumer_offsets topic (Confluent consumer docs). The leader of that partition serves as the group's coordinator. By default, that topic has 50 partitions and a replication factor of 3 (broker configs). When a group has no committed offset yet, auto.offset.reset sets the starting point: latest (the default) reads only new records, while earliest starts from the oldest retained record.

A rebalance reassigns a group's partitions across its members. Rebalances happen when a consumer joins the group, when a consumer leaves or fails, or when a subscribed topic gains new partitions. Kafka detects a failed consumer when it stops sending heartbeats for session.timeout.ms. Under group.protocol=consumer (described below), the broker setting group.consumer.session.timeout.ms replaces this client setting (broker configs). A consumer also leaves the group and triggers a rebalance when it fails to call poll within max.poll.interval.ms, for example because a batch takes too long to process (KafkaConsumer docs).

If C1 processed P1 up to offset 90 but committed only 87 before the rebalance, C2 reprocesses offsets 87 through 89. Because committing offsets after processing yields at-least-once processing, applications either make processing idempotent or use Kafka transactions. A Kafka transaction writes a consumer's output records and its offsets atomically, preventing a reprocessed batch from adding duplicate records to the output topic (design). Kafka 4.0 made a new rebalance protocol, KIP-848, generally available to speed up rebalances; consumers opt in by setting group.protocol=consumer (Kafka 4.0 release).

Retention and Log Compaction

Each topic's cleanup policy is configured via cleanup.policy (topic configs). The default policy, delete, removes old log segments by age using retention.ms (7 days by default) or by partition size using retention.bytes (no limit by default). Consumer reads do not affect retention: a record read by every consumer group stays on disk until it expires.

The alternative policy, compact, retains "at least the last known value for each message key" within a partition (design). Background threads remove older records that share the same key. A record with a key and a null value acts as a tombstone marking the key as deleted. Compaction removes prior records for that key and then deletes the tombstone itself after delete.retention.ms (1 day by default). A compacted topic acts like a table holding the latest state for each key, such as a booking's status. A new consumer can rebuild that table by reading the topic from the beginning.

Compaction preserves the offsets of surviving records: offset 4 stays offset 4, leaving gaps in the log. Kafka uses compaction on its internal __consumer_offsets topic because it needs to keep only the latest offset for each group and partition (broker configs, offsets.topic.segment.bytes).

KRaft: Cluster Metadata Without ZooKeeper

A Kafka cluster tracks metadata about existing topics, replica locations for each partition, current partition leaders, and ISR membership. Older versions of Kafka stored this metadata in a separate Apache ZooKeeper cluster. In KRaft mode, a small group of Kafka servers act as controllers that form a quorum and store metadata in an internal replicated log named __cluster_metadata. A majority of controllers must remain alive for the cluster to keep working (KRaft operations). One active controller manages leader elections. Operators typically run 3 or 5 controllers so the quorum survives one or two controller failures (KRaft operations).

ZooKeeper mode stores metadata in a second system with its own servers and operations, whereas KRaft mode keeps metadata inside the Kafka servers themselves. According to the upgrade notes, Kafka 4.0 supports only KRaft and removes ZooKeeper mode. A ZooKeeper-based cluster must migrate to KRaft on a 3.x release before it can upgrade to 4.0.

Kafka Connect and Schema Registry

Kafka Connect ships with Apache Kafka. It serves as "a tool for scalably and reliably streaming data between Apache Kafka and other systems." A source connector pulls data from an external system like a database and writes records into a topic. A sink connector reads records from a topic and writes them to an external system such as a search index or data warehouse. Connect runs in standalone mode for development and small deployments, or in distributed mode across a group of workers that share the connectors.

Schema Registry is not part of Apache Kafka; it is a separate Confluent service offered under the Confluent Community License (Schema Registry docs). It stores Avro, JSON Schema, or Protobuf record schemas for topics. Before sending a record, the producer's serializer registers the schema and embeds a short schema ID in the record instead of the full schema definition. The consumer's deserializer then uses that ID to fetch the schema. The registry rejects a new schema version that violates the topic's compatibility rule. Under the default BACKWARD rule, consumers using the new schema can read data written with the previous one (schema evolution).

Key idea. Because a new leader comes only from the in-sync replica set (ISR), a record written to every ISR replica survives a leader crash. With acks=1, a crash can lose an acknowledged record; with acks=all, the acknowledgment waits for every ISR replica. Pairing min.insync.replicas=2 with a replication factor of 3 makes the leader reject writes rather than accept them on a single copy.

The Kafka Exercise lesson runs a Kafka broker in Docker, creates a topic with 3 partitions, and shows which partition each key lands on.

Further Reading

• Introduction — Apache Kafka docs — records, topics, partitions, per-partition order, and the replication factor of 3.
• Design — Apache Kafka docs — replication, the ISR, committed records, unclean leader election, the idempotent producer, and log compaction.
• Producer configs, topic configs, and broker configs — acks, the default partitioner, min.insync.replicas, retention, replica.lag.time.max.ms, and the offsets topic.
• KafkaConsumer — Apache Kafka Javadoc — consumer groups, committed positions, and what triggers a rebalance.
• KIP-480 and KIP-794 — the sticky partitioner for keyless records and its uniform version.
• Apache Kafka 4.0 release announcement and upgrade notes — KRaft only, ZooKeeper removed, and KIP-848.
• Schema Registry — Confluent docs — schema formats, schema IDs, compatibility checks, and the license.