目录
Kafka集群选举
controller选举机制
Leader partition选举
leader partition自平衡
partition故障恢复机制
follower故障
leader故障
HW一致性保障
HW同步过程
Epoch
Kafka集群选举
1. 在多个broker中, 需要选举出一个broker, 担任controller. 由controller来管理整个集群中的分区和副本状态.
2. 在同一个topic下, 需要从多个partition中选举出一个leader节点, 来负责和客户端的交互, 优先写入, 同步给follower
controller选举机制
当集群kafka启动时, 所有的broker会尝试往zookeeper创建一个/controller的临时节点, 将自己的brokerid写入其中.zookeeper机制, 只会保证有一个broker写入成功, 成为controller.
由于是临时节点, zookeeper需要应用一直保持连接状态, 如果检测不到应用的心跳, zookeeper会删除临时节点, 同时会给监听该节点的客户端发送广播事件, 其他follower broker收到事件后, 会重新竞争controller.
客户端同时往zookeeper写入, 第一个写入成功(临时节点), 成为leader, 当leader挂掉, 临时节点被移除, 监听机制监听下线,重新竞争leader, 客户端也能监听最新leader
controller还会监听一些关键节点, 并推送给其他broker
- 监听Zookeeper中的/brokers/ids节点，感知Broker增减变化。
- 监听/brokers/topics，感知topic以及对应的partition的增减变化。
- 监听/admin/delete_topic节点，处理删除topic的动作。
Leader partition选举
一个topic的消息是由多个partition来存储的, 在用kafka-topics.sh创建topic时, 可以通过参数--partitions指定partition数量, 通过--replication-factors参数指定每个Partition有几个备份. 在一个partition的备份中, 会选举出一个leader, 来负责和客户端的交互, 以及同步数据给follower节点
partition参数:
- AR: Assigned Replicas, 分区中的所有副本, 包括存活和不存活
- ISR: 服务正常, 能够与leader保持通信的Follower副本
- OSR: 从ISR踢出的节点, 有问题或延迟过多的副本
选举过程: Replicas中越靠前越优先选取, 并且存在ISR, 也就是正常的服务, 被选为leader
leader partition自平衡
经过partiton选举, 可能造成大量leader存在同一个broker节点, 导致该broker压力明显大于其他broker, 影响集群性能. 为此，Kafka设计了Leader Partition自动平衡机制，当发现Leader分配不均衡时，自动进行Leader Partition调整。
kafka选举, 会把AR当中的第一个节点就应该是Leader节点。这种选举结果成为preferred election 理想选举结果。Controller会定期检测集群的Partition平衡情况，在开始检测时，Controller会依次检查所有的Broker。当发现这个Broker上的不平衡的Partition比例高于leader.imbalance.per.broker.percentage阈值时，会触发一次Leader Partiton的自平衡。也可以手动执行kafka-leader-election.sh脚本触发自平衡.
注意: Leader partition自平衡是一个很重的操作, 涉及大量消息转移和同步, 并且可能会丢消息. 在对性能要求较高的系统, 可以关闭自平衡, 设置auto.leader.rebalance.enable=false, 在业务不繁忙时候, 运维手动执行自平衡命令, 提高可用性.
partition故障恢复机制
当一组Partition中选举出了一个Leader节点后，这个Leader节点就会优先写入并保存Producer传递过来的消息，然后再同步给其他Follower。当Leader Partition所在的Broker服务发生宕机时，Kafka会触发Leader Partition的重新选举。Kafka为了保证消息能够在多个Parititon中保持数据同步，内部记录了两个关键参数
- Leo: 每个Partition的最后一个Offset
- HW: 一组Partiton中最小的LEO
partition每收到一条生产者发送的消息, LEO就会+1, follower从leader同步过来一条消息, LEO也会+1. follower从leader同步消息时, 会把自己的LEO传给leader, leader就会统计最小值, 同步给所有follower.
leader认为HW以前的消息, 也就是所有副本都存在的消息才是安全的, 可以被消费者拉取消费. 而HW之前的消息, 可能会丢失, 被认为不安全的.当一条消息发送到leader, 不会立刻让消费者感知, 而是等follower同步, 推进HW, 当HW大于消息时, 消费者才能消费,
follower故障
如果是Follower发生故障，这不会影响消息写入，只是少了一个备份
处理流程:
- 将故障的follower节点踢出ISR, 其他leader和follower正常工作
- 当故障follower恢复时, 不会立即加入ISR, 而且先同步消息, 把本地记录上一次HW, 并把大于HW的消息丢弃, 去leader同步消息
- 该follower的LEO大于partition的HW时, 假如ISR
leader故障
- 从ISR中选举出新的leader, 可能消息还未同步, 新leader的LEO小于老leader的LEO
- 其他follower会把大于HW的消息删除, 再从新leader同步消息
- 老leader恢复后, 会以follower身份加入, 也是先删大于HW, 再同步消息
HW一致性保障
HW同步过程
- follower先从leader拉取消息, 才能往leader上报LEO
- 当所有follower都上报后, leader才能计算HW值
- follower下一次拉取消息时, 才能更新HW
leader和follower的LEO是存在延迟的, 所以存在HW不一致问题. 当Leader切换时, HW不一致, follower按照自己的HW就行恢复数据, 可能造成数据不一致. Kafka设计Epoch来保证HW一致性
Epoch
Epoch由版本号和消息offset组成, 例如(1,100), 代表版本1, 一个单调递增的版本号, 当leader partiton发生变更时, 版本加一, 100表示当前partition写入第一条消息偏移量.
Broker会将这个epoch数据保存到内存中，并且会持久化到本地一个leader-epoch-checkpoint文件当中。leader-epoch-checkpoint会在所有Follower Partition中同步。当Leader Partition有变更时，新的Leader Partition就会读取这个Epoch记录，更新后添加自己的Epoch记录。
其他Follower Partition要更新数据时，不再靠自己记录的HW值判断拉取消息的起点, 而是根据最新的epoch来判断。
