您可以捐助,支持我们的公益事业。

1元 10元 50元





认证码:  验证码,看不清楚?请点击刷新验证码 必填



  求知 文章 文库 Lib 视频 iPerson 课程 认证 咨询 工具 讲座 Model Center   Code  
会员   
   
 
     
   
 订阅
  捐助
Kafka High Availability (中)
 
来源:技术世界 发布于:2017-9-20
   次浏览      
 

本文是系列文章的第三篇,第一篇 "Kafka设计解析(一)- Kafka背景及架构介绍"。第二篇 Kafka设计解析(二)- Kafka High Availability (上)

本文在上篇文章基础上,更加深入讲解了Kafka的HA机制,主要阐述了HA相关各种场景,如Broker failover,Controller failover,Topic创建/删除,Broker启动,Follower从Leader fetch数据等详细处理过程。

Controller对Broker Failure的处理过程

Controller在Zookeeper的/brokers/ids节点上注册Watch。一旦有Broker宕机(本文用宕机代表任何让Kafka认为其Broker die的情景,包括但不限于机器断电,网络不可用,GC导致的Stop The World,进程crash等),其在Zookeeper对应的Znode会自动被删除,Zookeeper会fire Controller注册的Watch,Controller即可获取最新的幸存的Broker列表。

Controller决定set_p,该集合包含了宕机的所有Broker上的所有Partition。

对set_p中的每一个Partition:

从/brokers/topics/[topic]/partitions/[partition]/state读取该Partition当前的ISR。

决定该Partition的新Leader。如果当前ISR中有至少一个Replica还幸存,则选择其中一个作为新Leader,新的ISR则包含当前ISR中所有幸存的Replica。否则选择该Partition中任意一个幸存的Replica作为新的Leader以及ISR(该场景下可能会有潜在的数据丢失)。如果该Partition的所有Replica都宕机了,则将新的Leader设置为-1。

将新的Leader,ISR和新的leader_epoch及controller_epoch写入/brokers/topics/[topic]/partitions/[partition]/state。注意,该操作只有Controller版本在3.1至3.3的过程中无变化时才会执行,否则跳转到3.1。

直接通过RPC向set_p相关的Broker发送LeaderAndISRRequest命令。Controller可以在一个RPC操作中发送多个命令从而提高效率。

Broker failover顺序图如下所示。

LeaderAndIsrRequest结构如下

LeaderAndIsrResponse结构如下

创建/删除Topic

Controller在Zookeeper的/brokers/topics节点上注册Watch,一旦某个Topic被创建或删除,则Controller会通过Watch得到新创建/删除的Topic的Partition/Replica分配。

对于删除Topic操作,Topic工具会将该Topic名字存于/admin/delete_topics。若delete.topic.enable为true,则Controller注册在/admin/delete_topics上的Watch被fire,Controller通过回调向对应的Broker发送StopReplicaRequest;若为false则Controller不会在/admin/delete_topics上注册Watch,也就不会对该事件作出反应,此时Topic操作只被记录而不会被执行。

对于创建Topic操作,Controller从/brokers/ids读取当前所有可用的Broker列表,对于set_p中的每一个Partition:

从分配给该Partition的所有Replica(称为AR)中任选一个可用的Broker作为新的Leader,并将AR设置为新的ISR(因为该Topic是新创建的,所以AR中所有的Replica都没有数据,可认为它们都是同步的,也即都在ISR中,任意一个Replica都可作为Leader)

将新的Leader和ISR写入/brokers/topics/[topic]/partitions/[partition]

直接通过RPC向相关的Broker发送LeaderAndISRRequest。

创建Topic顺序图如下所示。

Broker响应请求流程

Broker通过kafka.network.SocketServer及相关模块接受各种请求并作出响应。整个网络通信模块基于Java NIO开发,并采用Reactor模式,其中包含1个Acceptor负责接受客户请求,N个Processor负责读写数据,M个Handler处理业务逻辑。

Acceptor的主要职责是监听并接受客户端(请求发起方,包括但不限于Producer,Consumer,Controller,Admin Tool)的连接请求,并建立和客户端的数据传输通道,然后为该客户端指定一个Processor,至此它对该客户端该次请求的任务就结束了,它可以去响应下一个客户端的连接请求了。其核心代码如下。

Processor主要负责从客户端读取数据并将响应返回给客户端,它本身并不处理具体的业务逻辑,并且其内部维护了一个队列来保存分配给它的所有SocketChannel。Processor的run方法会循环从队列中取出新的SocketChannel并将其SelectionKey.OP_READ注册到selector上,然后循环处理已就绪的读(请求)和写(响应)。Processor读取完数据后,将其封装成Request对象并将其交给RequestChannel。

RequestChannel是Processor和KafkaRequestHandler交换数据的地方,它包含一个队列requestQueue用来存放Processor加入的Request,KafkaRequestHandler会从里面取出Request来处理;同时它还包含一个respondQueue,用来存放KafkaRequestHandler处理完Request后返还给客户端的Response。

Processor会通过processNewResponses方法依次将requestChannel中responseQueue保存的Response取出,并将对应的SelectionKey.OP_WRITE事件注册到selector上。当selector的select方法返回时,对检测到的可写通道,调用write方法将Response返回给客户端。

KafkaRequestHandler循环从RequestChannel中取Request并交给kafka.server.KafkaApis处理具体的业务逻辑。

LeaderAndIsrRequest响应过程

对于收到的LeaderAndIsrRequest,Broker主要通过ReplicaManager的becomeLeaderOrFollower处理,流程如下:

若请求中controllerEpoch小于当前最新的controllerEpoch,则直接返回ErrorMapping.StaleControllerEpochCode。

对于请求中partitionStateInfos中的每一个元素,即((topic, partitionId), partitionStateInfo):

若partitionStateInfo中的leader epoch大于当前ReplicManager中存储的(topic, partitionId)对应的partition的leader epoch,则:

若当前brokerid(或者说replica id)在partitionStateInfo中,则将该partition及partitionStateInfo存入一个名为partitionState的HashMap中

否则说明该Broker不在该Partition分配的Replica list中,将该信息记录于log中

否则将相应的Error code(ErrorMapping.StaleLeaderEpochCode)存入Response中

筛选出partitionState中Leader与当前Broker ID相等的所有记录存入partitionsTobeLeader中,其它记录存入partitionsToBeFollower中。

若partitionsTobeLeader不为空,则对其执行makeLeaders方。

若partitionsToBeFollower不为空,则对其执行makeFollowers方法。

若highwatermak线程还未启动,则将其启动,并将hwThreadInitialized设为true。

关闭所有Idle状态的Fetcher。

LeaderAndIsrRequest处理过程如下图所示

Broker启动过程

Broker启动后首先根据其ID在Zookeeper的/brokers/idszonde下创建临时子节点(Ephemeral node),创建成功后Controller的ReplicaStateMachine注册其上的Broker Change Watch会被fire,从而通过回调KafkaController.onBrokerStartup方法完成以下步骤:

1. 向所有新启动的Broker发送UpdateMetadataRequest,其定义如下

2. 将新启动的Broker上的所有Replica设置为OnlineReplica状态,同时这些Broker会为这些Partition启动high watermark线程。

3. 通过partitionStateMachine触发OnlinePartitionStateChange。

Controller Failover

Controller也需要Failover。每个Broker都会在Controller Path (/controller)上注册一个Watch。当前Controller失败时,对应的Controller Path会自动消失(因为它是Ephemeral Node),此时该Watch被fire,所有“活”着的Broker都会去竞选成为新的Controller(创建新的Controller Path),但是只会有一个竞选成功(这点由Zookeeper保证)。竞选成功者即为新的Leader,竞选失败者则重新在新的Controller Path上注册Watch。因为Zookeeper的Watch是一次性的,被fire一次之后即失效,所以需要重新注册。

Broker成功竞选为新Controller后会触发KafkaController.onControllerFailover方法,并在该方法中完成如下操作:

读取并增加Controller Epoch。

在ReassignedPartitions Patch (/admin/reassign_partitions)上注册Watch。

在PreferredReplicaElection Path (/admin/preferred_replica_election)上注册Watch。

通过partitionStateMachine在Broker Topics Patch (/brokers/topics)上注册Watch。

若delete.topic.enable设置为true(默认值是false),则partitionStateMachine在Delete Topic Patch (/admin/delete_topics)上注册Watch。

通过replicaStateMachine在Broker Ids Patch (/brokers/ids)上注册Watch。

初始化ControllerContext对象,设置当前所有Topic,“活”着的Broker列表,所有Partition的Leader及ISR等。

启动replicaStateMachine和partitionStateMachine。

将brokerState状态设置为RunningAsController。

将每个Partition的Leadership信息发送给所有“活”着的Broker。

若auto.leader.rebalance.enable配置为true(默认值是true),则启动partition-rebalance线程。

若delete.topic.enable设置为true且Delete Topic Patch(/admin/delete_topics)中有值,则删除相应的Topic。

Partition重新分配

管理工具发出重新分配Partition请求后,会将相应信息写到/admin/reassign_partitions上,而该操作会触发ReassignedPartitionsIsrChangeListener,从而通过执行回调函数KafkaController.onPartitionReassignment来完成以下操作:

将Zookeeper中的AR(Current Assigned Replicas)更新为OAR(Original list of replicas for partition) + RAR(Reassigned replicas)。

强制更新Zookeeper中的leader epoch,向AR中的每个Replica发送LeaderAndIsrRequest。

将RAR - OAR中的Replica设置为NewReplica状态。

等待直到RAR中所有的Replica都与其Leader同步。

将RAR中所有的Replica都设置为OnlineReplica状态。

将Cache中的AR设置为RAR。

若Leader不在RAR中,则从RAR中重新选举出一个新的Leader并发送LeaderAndIsrRequest。若新的Leader不是从RAR中选举而出,则还要增加Zookeeper中的leader epoch。

将OAR - RAR中的所有Replica设置为OfflineReplica状态,该过程包含两部分。第一,将Zookeeper上ISR中的OAR - RAR移除并向Leader发送LeaderAndIsrRequest从而通知这些Replica已经从ISR中移除;第二,向OAR - RAR中的Replica发送StopReplicaRequest从而停止不再分配给该Partition的Replica。

将OAR - RAR中的所有Replica设置为NonExistentReplica状态从而将其从磁盘上删除。

将Zookeeper中的AR设置为RAR。

删除/admin/reassign_partition。

注意:最后一步才将Zookeeper中的AR更新,因为这是唯一一个持久存储AR的地方,如果Controller在这一步之前crash,新的Controller仍然能够继续完成该过程。

以下是Partition重新分配的案例,OAR = {1,2,3},RAR = {4,5,6},Partition重新分配过程中Zookeeper中的AR和Leader/ISR路径如下

Follower从Leader Fetch数据

Follower通过向Leader发送FetchRequest获取消息,FetchRequest结构如下

从FetchRequest的结构可以看出,每个Fetch请求都要指定最大等待时间和最小获取字节数,以及由TopicAndPartition和PartitionFetchInfo构成的Map。实际上,Follower从Leader数据和Consumer从Broker Fetch数据,都是通过FetchRequest请求完成,所以在FetchRequest结构中,其中一个字段是clientID,并且其默认值是ConsumerConfig.DefaultClientId。

Leader收到Fetch请求后,Kafka通过KafkaApis.handleFetchRequest响应该请求,响应过程如下:

replicaManager根据请求读出数据存入dataRead中。

如果该请求来自Follower则更新其相应的LEO(log end offset)以及相应Partition的High Watermark

根据dataRead算出可读消息长度(单位为字节)并存入bytesReadable中。

满足下面4个条件中的1个,则立即将相应的数据返回

Fetch请求不希望等待,即fetchRequest.macWait <= 0

Fetch请求不要求一定能取到消息,即fetchRequest.numPartitions <= 0,也即requestInfo为空

有足够的数据可供返回,即bytesReadable >= fetchRequest.minBytes

读取数据时发生异常

若不满足以上4个条件,FetchRequest将不会立即返回,并将该请求封装成DelayedFetch。检查该DeplayedFetch是否满足,若满足则返回请求,否则将该请求加入Watch列表

Leader通过以FetchResponse的形式将消息返回给Follower,FetchResponse结构如下

   
次浏览       
相关文章

基于EA的数据库建模
数据流建模(EA指南)
“数据湖”:概念、特征、架构与案例
在线商城数据库系统设计 思路+效果
 
相关文档

Greenplum数据库基础培训
MySQL5.1性能优化方案
某电商数据中台架构实践
MySQL高扩展架构设计
相关课程

数据治理、数据架构及数据标准
MongoDB实战课程
并发、大容量、高性能数据库设计与优化
PostgreSQL数据库实战培训