- Differences between V1 and V2
- Paths stored in Zookeeper
- Key data structures
- Key algorithms
- Zookeeper listeners ONLY on the leader
- Zookeeper listeners on all brokers
- Configuration parameters
- Broker startup
- Leader election
- State change events
- State change operations
- Admin commands
- Handling produce requests
- Message replication
Differences between V1 and V2
This detailed design differs from the original detailed design in the following areas -
- This design aims to remove split-brain and herd effect issues in the V1 design. A partition has only one brain (on the leader) and all brokers only respond to state changes that are meant for them (as decided by the leader).
- The state machine in this design is completely controlled only by the leader for each partition. Each follower changes its state only based on such a request from the leader for a particular partition. Leader co-ordinated state machine allows central state machine verification and allows it to fail fast.
- This design introduces an epoch or generation id per partition, which is a non-decreasing value for a partition. The epoch increments when the leader for a partition changes.
- This design handles delete partition or delete topic state changes for dead brokers by queuing up state change requests for a broker in Zookeeper.
- This design scales better wrt to number of ZK watches, since it registers fewer watches compared to V1. The motivation is to be able to reduce the load on ZK when the Kafka cluster grows to thousands of partitions. For example, if we have a cluster of 3 brokers hosting 1000 topics with 3 partitions each, the V1 design requires registering 15000 watches. The V2 design requires registering 3000 watches.
- This design ensures that leader change ZK notifications are not queued up on any other notifications and can happen instantaneously.
- This design allows explicit monitoring of
- the entire lifecycle of a state change -
- leader, broker id 0, requested start-replica for topic foo partition 0, to broker id 1, at epoch 10
- leader, broker id 0, requested start-replica for topic foo partition 0, to broker id 2, at epoch 10
- follower, broker id 1, received start-replica for topic foo partition 0, from leader 0, at epoch 10
- follower, broker id 2, received start-replica for topic foo partition 0, from leader 0, at epoch 10
- follower, broker id 1, completed start-replica for topic foo partition 0, request from leader 0, at epoch 10
- follower, broker id 2, completed start-replica for topic foo partition 0, request from leader 0, at epoch 10
- the backup of state change requests, on slow followers
- the entire lifecycle of a state change -
Paths stored in Zookeeper
Notation: When an element in a path is denoted [xyz], that means that the value of xyz is not fixed and there is in fact a znode for each possible value of xyz. For example /topics/[topic] would be a directory named /topics containing a directory for each topic name. An arrow -> is used to indicate the contents of a znode. For example /hello -> world would indicate a znode /hello containing the value "world". A path is persistent unless it’s marked as ephemeral.
We store the following paths in Zookeeper:
- Stores the information of all live brokers.
- Stores for each partition, a list of the currently assigned replicas. For each replica, we store the id of the broker to which the replica is assigned. The first replica is the preferred replica. Note that for a given partition, there is at most 1 replica on a broker. Therefore, the broker id can be used as the replica id
- Stores the id of the replica that’s the current leader of this partition
- Stores the id of the set of replicas that are in-sync with the leader
- This path is used when we want to reassign some partitions to a different set of brokers. For each partition to be reassigned, it stores a list of new replicas and their corresponding assigned brokers. This path is created by an administrative process and is automatically removed once the partition has been moved successfully
- This path is used by the leader of a partition to enqueue state change requests to the follower replicas. The various state change requests include start replica, close replica. This path is created by the add brokers admin command. This path is only deleted by the remove brokers admin command. The purpose of making this path persistent is to cleanly handle state changes like delete topic and reassign partitions even when a broker is temporarily unavailable (for example, being bounced).
Key data structures
Every broker stores a list of partitions and replicas assigned to it. The current leader of a partition further maintains 3 sets: AR, ISR, CUR and RAR, which correspond to the set of replicas that are assigned to the partition, in-sync with the leader, catching up with the leader, and being reassigned to other brokers. Normally, ISR is a subset of AR and AR = ISR + CUR. The leader of a partition maintains a commitQ and uses it to buffer all produce requests to be committed. For each replica assigned to a broker, the broker periodically stores its HW in a checkpoint file.
Zookeeper listeners ONLY on the leader
- Partition-reassigned listener:
- child change on /brokers/partitions_reassigned
- child change on /brokers/partitions_reassigned/[topic]
Zookeeper listeners on all brokers
- Leader-change listener: value change on /brokers/topics/[topic]/[partition_id]/leader
- State-change listener: child change on /brokers/state/[broker_id]
- LeaderElectionWaitTime: controls the maximum amount of time that we wait during leader election.
- KeepInSyncTime: controls the maximum amount of time that a leader waits before dropping a follower from the in-sync replica set.
Each time a broker starts up, it calls brokerStartup() and the algorithms are described below
State change events
On every broker
This leader change listener is registered on every broker hosting a partition p. Each time it is triggered, the following procedure is executed -
On State change
Each broker has a ZK path that it listens to for state change requests from the leader
On the leader
On reassignment of partitions
Each time a partition reassigned event is triggered on the leader, it calls onPartitionReassigned()
State change communication
The leader uses this API to communicate a state change request to the followers
State change operations
This state change is requested by the leader or the admin command for a new replica assignment
This state change is requested by the leader when a topic or partition is deleted or moved to another broker
This state change is requested by the leader when the leader for a replica changes
This state change is done by the new leader
This section describes the algorithms for various admin commands like create/delete topic, add/remove partition.
The admin commands does the following while creating a new topic
Add partition to existing topic
Remove partition for existing topic
Handling produce requests
Produce request handler on the leader
Commit thread on the leader
Follower fetching from leader
A follower keeps sending ReplicaFetcherRequests to the leader. The process at the leader and the follower are described below -
At the leader
At the follower
ReplicaFetcherThread for Replica r: