Replication: Advanced

A clean election picks the new leader from the in-sync replicas, so the new leader holds every record the old one had committed. That rule has a gap. The ISR can shrink to the leader alone when every follower falls behind, and if that leader then dies, no replica is in sync with it. The controller has 2 options: leave the partition without a leader until the dead broker returns, or make a replica that is missing records the leader. The setting that picks between them is unclean.leader.election.enable.

Take partition 3 of orders, replicated on brokers 0, 1 and 2 with broker 0 leading. Both followers sit on a rack whose switch starts dropping packets, their fetches stall, and after replica.lag.time.max.ms the leader removes them from the ISR. Writes keep landing: min.insync.replicas is 1, the ISR is {0}, and the high watermark advances with every record. The log end offsets drift apart.

With the setting at its default of false, the controller leaves partition 3 with no leader. Writes to it fail, readers on it get nothing, and the other partitions of orders carry on. When broker 0 comes back with its log intact, it resumes as leader at offset 100, the followers catch up, and every acknowledged record is still there. The price is that this partition stays offline for as long as broker 0 is down.

With the setting at true, the controller makes a replica from outside the ISR the leader. It never sees the replicas' log end offsets, so it does not look for the freshest one. It takes the first live broker in the partition's replica list, here [0, 1, 2], which is broker 1. Broker 1 leads from its own log end, offset 61, and the ISR becomes {1}. Records 61 through 99, all acknowledged, now exist only on broker 0's powered-off disk. Had the list been [0, 2, 1], broker 2 would lead from offset 58 and records 58 to 99 would be lost, 3 more than here, while broker 1 sat holding them. New writes arrive at once and the leader assigns them offsets starting at 61. Offset 80 now names a record written after the failure, and any reader that already consumed the old record 80 holds a value the cluster no longer has.

The setting is a topic config and overrides the broker default. A topic of page-view counts can run with it on, since a gap of 40 records costs a dashboard a blip. A topic of payments runs with it off and accepts the outage. You can also leave it off everywhere and trigger a single unclean election by hand when the wait has gone on too long: kafka-leader-election.sh --bootstrap-server b1:9092 --election-type UNCLEAN --topic orders --partition 3. That choice is yours at the moment, rather than the controller's by rule.

Each node declares process.roles: controller, broker, or broker,controller for both on one process. Controllers form the quorum; brokers hold topic partitions. Running both on 1 process suits a laptop, and a production cluster keeps them on separate machines so that a broker drowning in fetches never delays an election.

A broker's file names the same bootstrap list under process.roles=broker, which is how it finds the quorum. Each node is formatted once with kafka-storage.sh format, which stamps the cluster id into its log directory, and a controller formatted with --initial-controllers joins as a voter from the start.

The quorum's state is a topic named __cluster_metadata with 1 partition, held by every controller. One controller is its leader, called the active controller, and the others replicate it as followers, with the same fetch loop a topic partition uses. A record in it counts as committed once a majority of the voters have written it: 2 of 3, or 3 of 5. Each record describes one change: a topic created, a partition's ISR shrinking, a broker registering, a leader changing. Replaying the log from the start rebuilds the whole cluster's metadata, which is what a controller does when it boots, and it snapshots the log periodically so that the replay stays short.

Each broker sends a heartbeat to the active controller on a fixed interval, and when the controller has heard nothing for longer than the session timeout it fences the broker: it writes a record marking it fenced, then one PartitionChangeRecord per partition that broker led, naming a new leader from each ISR. Those records commit once a majority of controllers hold them. Each broker learns by fetching: it runs a fetch against the metadata log, exactly as a follower fetches a partition, and applies every new record it reads. A broker finds out it has become leader of orders-3 by reading that fact from the log.

kafka-metadata-quorum.sh --bootstrap-server b1:9092 describe --status prints the active controller and how far each voter has replicated:

kafka-reassign-partitions.sh runs in 3 steps. --generate takes a list of topics and a --broker-list and proposes a spread, --execute submits a reassignment file, and --verify reports which partitions have finished and clears the throttles it set. The file names the target replica set per partition:

The first entry of replicas is the preferred leader, so the file also decides who leads once the move finishes. log_dirs can name a specific disk on the broker instead of any, which is how a partition is moved from a full disk to an empty one on the same machine.

For orders-3, moving from [0, 1, 2] to [1, 2, 4], the controller records broker 4 as an adding replica and broker 0 as a removing replica. The partition's replica set becomes [0, 1, 2, 4] for the duration. Broker 4 opens a fetch against the leader at the leader's log start offset and copies every segment, while the ISR stays {0, 1, 2} and writes are unaffected. When broker 4 is caught up, the leader adds it to the ISR. The controller then drops broker 0 from the replica set and the ISR, moves leadership to broker 1 because broker 0 was leading, and broker 0 deletes its copy. While broker 4 copies and sits outside the ISR, no write waits for it. Once it joins the ISR it counts like any other in-sync replica, and a write that is acknowledged only when every in-sync replica has it waits for broker 4 too.

A moving replica reads the whole partition over the network and writes it to a new disk. A 400 GB partition is 400 GB of reads from the leader's disk, 400 GB across the network and 400 GB of writes, on top of the traffic the topic already carries. Unthrottled, that copy competes with the leader's followers and with the writers sending to it. The tool takes --throttle in bytes per second, which sets leader.replication.throttled.rate and follower.replication.throttled.rate on the brokers involved and lists only the moving replicas as throttled on the topic, so in-sync followers are never slowed.

The copy must also outrun the writers. A partition taking in 20 MB/s under a 50 MB/s throttle closes the gap at 30 MB/s, and a throttle below the write rate never finishes. Running --verify after the move clears the throttle settings. Skipping it leaves the throttled-replica lists on the topic and the rate settings on the brokers in place, so replication of those replicas stays capped at the old rate, and any later move that reuses them crawls at that rate too. On a topic with tiered storage, the new replica copies only the local segments and reads the remote ones from the object store when asked, which makes the move as cheap as the local retention.

A move in flight can be stopped with --cancel, which removes the adding replicas and leaves the originals in place; the copied bytes are discarded. Submitting a second file for a partition already moving replaces its target. A move is also how a replication factor is raised, by naming 4 brokers where there were 3, or lowered, by naming 2.

Replication protects a partition against a broker dying. It does nothing when the whole cluster goes: a region's power, a bad configuration push, a disk corruption that every replica copied. For that, a second cluster in another place holds a copy of the topic, written by a mirror. Apache Kafka ships one as MirrorMaker 2, started with connect-mirror-maker.sh and a properties file.

Topic orders on east appears on west as east.orders, with the same partition count and each record in the same partition number it had at the source. The prefix is the source cluster's alias and does 2 jobs: a reader on west can tell local records from mirrored ones, and a mirror running the other way, west to east, skips any topic whose prefix names east, so 2 clusters can mirror each other without a record circling forever. Topic configs such as retention are copied too, so the mirror's log is cleaned by the same rules as the source.

A reader of orders on east keeps its position as a source offset, say 1202. On west that number means nothing, since offset 1202 of east.orders is a different record, if it exists at all. The mirror solves this by writing offset syncs: a record, on a topic of its own named mm2-offset-syncs.west.internal, pairing a source offset with the target offset the same record received, written whenever the pair has moved by offset.lag.max records. That topic lives on the source cluster by default, which is useless when the source is the cluster that died, so a mirror built for failover sets offset-syncs.topic.location=target to keep it on west. To fail over, the reader finds the latest sync whose source offset is at or below its own, and starts from that sync's target offset.

Readers that keep their positions on the cluster itself can have the mirror translate them in bulk; that mechanism belongs to a later lesson on consumer groups.

The broker needs replica.selector.class set to the rack-aware selector and a broker.rack naming its zone, which rack-aware placement already requires. The reader sets client.rack to the zone it runs in. On the reader's first fetch, the leader answers with no records and a preferred read replica, the follower whose rack matches; the reader sends its next fetches there.

A follower serves only up to its own high watermark, and it learns the leader's high watermark one fetch late: the leader advances it when every in-sync replica has fetched a record, and tells the followers in the response to their next fetch. A record committed at the leader at 10:00:00.000 becomes readable from the leader at once and from the follower after the follower's next round trip, which is bounded by replica.fetch.wait.max.ms, 500 ms by default, plus the network.

Two readers of the same partition in different zones therefore see each record at different moments, around 140 ms apart in this table and up to half a second in a quiet partition. They never see different records at an offset, and never a record the leader has not committed, because the follower's high watermark never passes the leader's. The order inside the partition is unchanged. What changes is that latest means something different for each reader: the end of the log as the follower knows it.

The leader picks the follower only from the ISR, so a reader is never pointed at a replica that fell behind. If the chosen follower then drops out of the ISR or dies, the reader's fetch fails, it goes back to the leader, and the leader names a new preferred replica or serves the reader itself. Every metadata.max.age.ms, 5 minutes by default, the reader asks the leader again, which lets the leader move it after a reassignment puts a replica in its zone. If no replica shares the reader's rack, it reads from the leader as before. Writes are untouched: a follower never accepts a record, so the reader's zone decides where its reads go and nothing about where writes go.