0%

kafka之ReplicaManager

Kafka的副本机制

Kafka中主题的每个Partition有一个预写式日志文件,每个Partition都由一系列有序的、不可变的消息组成,这些消息被连续的追加到Partition中,Partition中的每个消息都有一个连续的序列号叫做offset, 确定它在分区日志中唯一的位置。

20220409143015

Kafka每个topic的partition有N个副本,其中N是topic的复制因子。Kafka通过多副本机制实现故障自动转移,当Kafka集群中一个Broker失效情况下仍然保证服务可用。在Kafka中发生复制时确保partition的预写式日志有序地写到其他节点上。N个replicas中。其中一个replica为leader,其他都为follower,leader处理partition的所有读写请求,与此同时,follower会被动定期地去复制leader上的数据。

20220409143223

Kafka提供了数据复制算法保证,如果leader发生故障或挂掉,一个新leader被选举并被接受客户端的消息成功写入。Kafka确保从同步副本列表中选举一个副本为leader,或者说follower追赶leader数据。leader负责维护和跟踪ISR(In-Sync Replicas的缩写,表示副本同步队列,具体可参考下节)中所有follower滞后的状态。当producer发送一条消息到broker后,leader写入消息并复制到所有follower。消息提交之后才被成功复制到所有的同步副本。消息复制延迟受最慢的follower限制,重要的是快速检测慢副本,如果follower“落后”太多或者失效,leader将会把它从ISR中删除。

leader和follower的角色区分,也主要是ReplicaManager来实现。具体地讲:

leader

  1. leader会接受client的读取请求和写入请求。
  2. leader需要接受follwer抓取message的请求,返回message给follower
  3. leader需要维护ISR(in-sync replicas)列表。“保持同步”的含义有些复杂,0.9之前版本对这个概念的定义与0.9不同,详情参见KIP-16 - Automated Replica Lag Tuning。0.9版本,broker的参数replica.lag.time.max.ms用来指定ISR的定义,如果leader在这么长时间没收到follower的拉取请求,或者在这么长时间内,follower没有fetch到leader的log end offset,就会被leader从ISR中移除。ISR是个很重要的指标,controller选取partition的leader replica时会使用它,因此leader选取ISR后会把结果记到Zookeeper上。
  4. leader需要维护high watermark。high watermark以下的消息就是所有ISR列表里的replica都已经读取的消息(注意,并不是所有replica都一定有这些消息,而只是ISR里的那些才肯定会有)。因此leader会根据follower拉取数据时提供的offset和ISR列表,决定HW,并且在返回给follower的请求中附带最新的HW。

follower

  1. follower需要不停地去leader处拉取最新的log
  2. follower需要根据leader在fetch reponse中提供的HW,更新自己本地保存的leader的HW信息。在它过行leader或follower转变时,会用到这个HW。

副本同步队列ISR

所谓同步,必须满足如下两个条件:

  1. 副本节点必须能与zookeeper保持会话(心跳机制)
  2. 副本能复制leader上的所有写操作,并且不能落后太多。(卡住或滞后的副本控制是由 replica.lag.time.max.ms 配置)

所有的副本(replicas)统称为Assigned Replicas,即AR。ISR是AR中的一个子集,由leader维护ISR列表,follower从leader同步数据有一些延迟。任意一个超过阈值都会把follower剔除出ISR, 存入OSR(Outof-Sync Replicas)列表,新加入的follower也会先存放在OSR中。AR=ISR+OSR。

上一节中的HW俗称高水位,是HighWatermark的缩写,取一个partition对应的ISR中最小的LEO作为HW,consumer最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息,consumer不能立刻消费,leader会等待该消息被所有ISR中的replicas同步后更新HW,此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效,该消息仍然可以从新选举的leader中获取。对于来自内部broKer的读取请求,没有HW的限制。

下图详细的说明了当producer生产消息至broker后,ISR以及HW和LEO的流转过程:

20220409143654

由此可见,Kafka的复制机制既不是完全的同步复制,也不是单纯的异步复制。事实上,同步复制要求所有能工作的follower都复制完,这条消息才会被commit,这种复制方式极大的影响了吞吐率。而异步复制方式下,follower异步的从leader复制数据,数据只要被leader写入log就被认为已经commit,这种情况下如果follower都还没有复制完,落后于leader时,突然leader宕机,则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。

Welcome to my other publishing channels