取本地数据_深入理解Kafka服务端之Follower副本如何同步Leader副本的数据
生活随笔
收集整理的這篇文章主要介紹了
取本地数据_深入理解Kafka服务端之Follower副本如何同步Leader副本的数据
小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.
一、場景分析Kafka采用的是主寫主讀的方式，即客戶端的讀寫請求都由分區(qū)的Leader副本處理，那么Follower副本要想保證和Leader副本數(shù)據(jù)一致，就需要不斷地從Leader副本拉取消息來進行同步。由于同一個分區(qū)的Leader副本和Follower副本分布在不同的節(jié)點上，所以同步的過程可以簡單概括為：Follower副本所在節(jié)點封裝拉取數(shù)據(jù)的請求并發(fā)送給Leader副本所在節(jié)點 → Leader副本所在節(jié)點接收拉取數(shù)據(jù)的請求并進行處理，然后返回響應(yīng) → Follower副本所在節(jié)點接收到返回的響應(yīng)并進行處理。這個過程中封裝拉取請求和處理返回的響應(yīng)是Follower副本所在節(jié)點的一個單獨的線程完成的。二、圖示說明leader 副本將數(shù)據(jù)寫入本地磁盤 leader?副本更新 LEO follower 副本發(fā)送同步數(shù)據(jù)請求，攜帶自身的 LEO leader 副本更新本地保存的其它副本的 LEO leader 副本嘗試更新 ISR 列表 leader?副本更新 HW leader 副本給 follower 副本返回數(shù)據(jù)，攜帶 leader 副本的 HW 值 follower 副本接收響應(yīng)并寫入數(shù)據(jù)，更新自身 LEO follower 副本更新本地的 HW 值 leader 副本將數(shù)據(jù)寫入本地磁盤 leader?副本更新 LEO follower 副本發(fā)送同步數(shù)據(jù)請求，攜帶自身的 LEO leader 副本更新本地保存的其它副本的 LEO leader 副本嘗試更新 ISR 列表 leader?副本更新 HW leader 副本給 follower 副本返回數(shù)據(jù)，攜帶 leader 副本的 HW 值 follower 副本接收響應(yīng)并寫入數(shù)據(jù)，更新自身 LEO follower 副本更新本地的 HW 值
????假設(shè)某主題只有1個分區(qū)，該分區(qū)有兩個副本：Leader 副本在 Broker1 上，Follower 副本在 Broker2 上，其 Leader 副本寫入數(shù)據(jù)和 Follower 副本同步數(shù)據(jù)的流程如下圖：
三、源碼分析Kafka分區(qū)的Leader副本接收客戶端生產(chǎn)的數(shù)據(jù)，寫入本地存儲；然后Follower副本拉取數(shù)據(jù)寫入本地存儲，并更新一系列關(guān)鍵的偏移量。整個流程比較復(fù)雜，這里先通過一個簡單的方法調(diào)用流程來看一下這個過程：1.leader?副本將數(shù)據(jù)寫入本地磁盤 KafkaApis.handleProduceRequest(){ replicaManager.appendRecords(){ appendToLocalLog(){ Partition.appendRecordsToLeader(){ Log.appendAsLeader(){ Log.append(){ //通過LogSegment.append()方法寫入磁盤 LogSegment.append() } } } } } }2.leader 副本更新LEO KafkaApis.handleProduceRequest(){ replicaManager.appendRecords(){ appendToLocalLog(){ Partition.appendRecordsToLeader(){ Log.appendAsLeader(){ Log.append(){ //更新Leader副本的LEO值 updateLogEndOffset(appendInfo.lastOffset + 1) } } } } } }3.follower?副本同步數(shù)據(jù)，攜帶自身的LEO AbstractFetchThread.doWork(){ maybeFetch(){ buildFetch(fetchStates){ //這里的fetchState.fetchOffset 就是Follower副本的LEO值 builder.add(topicPartition, new FetchRequest.PartitionData( fetchState.fetchOffset, logStartOffset, fetchSize, Optional.of(fetchState.currentLeaderEpoch))) } } }4.leader 副本更新本地保存的Follower副本的LEO ReplicaManager.fetchMessages(){ //獲取讀取結(jié)果 val logReadResults = readFromLog(){ if (isFromFollower) updateFollowerLogReadResults(replicaId, result){ //TODO 更新leader保存的各個follower副本的LEO partition.updateReplicaLogReadResult(replica, readResult){ //TODO 最終更新所有的replica的LEO的值 replica.updateLogReadResult(logReadResult){ //更新LEO對象 logEndOffsetMetadata = logReadResult.info.fetchOffsetMetadata } } } } }5.leader 副本嘗試更新ISR列表 ReplicaManager.fetchMessages(){ //獲取讀取結(jié)果 val logReadResults = readFromLog(){ if (isFromFollower) updateFollowerLogReadResults(replicaId, result){ //TODO 嘗試更新ISR列表 val leaderHWIncremented = maybeExpandIsr(replicaId, logReadResult){ //更新ISR列表 updateIsr(newInSyncReplicas) } } } }6.leader 副本更新HW ReplicaManager.fetchMessages(){ //獲取讀取結(jié)果 val logReadResults = readFromLog(){ if (isFromFollower) updateFollowerLogReadResults(replicaId, result){ //TODO 嘗試更新ISR列表及Leader副本的HW val leaderHWIncremented = maybeExpandIsr(replicaId, logReadResult){ //TODO 嘗試更新leader的HW maybeIncrementLeaderHW(leaderReplica, logReadResult.fetchTimeMs){ //取ISR列表中副本的最小的LEO作為新的HW val newHighWatermark = allLogEndOffsets.min(new LogOffsetMetadata.OffsetOrdering) //獲取舊的HW val oldHighWatermark = leaderReplica.highWatermark //如果新的HW值大于舊的HW值，就更新 if (oldHighWatermark.messageOffset < newHighWatermark.messageOffset || (oldHighWatermark.messageOffset == newHighWatermark.messageOffset && oldHighWatermark.onOlderSegment(newHighWatermark))) { //更新 Leader 副本的 HW????????????????????????????leaderReplica.highWatermark?=?newHighWatermark } } } } } }7.leader 副本給 follower副本 返回數(shù)據(jù)，攜帶leader 副本的 HW 值 ReplicaManager.fetchMessages(){ //獲取讀取結(jié)果 val logReadResults = readFromLog(){ readFromLocalLog(){ read(){ val readInfo = partition.readRecords(){ //獲取Leader Replica的高水位 val initialHighWatermark = localReplica.highWatermark.messageOffset } } } } }8.follower 副本寫入數(shù)據(jù)，更新自身LEO、 ReplicaFetcherThread.processPartitionData(){ partition.appendRecordsToFollowerOrFutureReplica(records, isFuture = false){ doAppendRecordsToFollowerOrFutureReplica(){ Log.appendAsFollower(){ Log.append(){ //更新Follower副本的LEO值 updateLogEndOffset(appendInfo.lastOffset + 1) } } } } }9.follower 副本更新本地的 HW 值 ReplicaFetcherThread.processPartitionData(){ //根據(jù)leader返回的HW，更新Follower本地的HW：取Follower本地LEO 和 Leader HW 的較小值 val followerHighWatermark = replica.logEndOffset.min(partitionData.highWatermark) //TODO 更新Follower副本的 HW 對象 replica.highWatermark = new LogOffsetMetadata(followerHighWatermark)??}注意：對于HW，Leader 副本和 Follower 副本只保存自身的
對于LEO，Follower 副本只保存自身的，但是 Leader 副本除了保存自身的外，還會保存所有 Follower 副本的 LEO 值
無論是Leader副本所在節(jié)點，還是Follower副本所在節(jié)點，分區(qū)對應(yīng)的Partition 對象都會保存所有的副本對象，但是只有本地副本對象有對應(yīng)的日志文件
整個數(shù)據(jù)寫入及同步的過程分為九個步驟：
? ?下面具體分析這幾個步驟。第一、二步在分析日志對象的寫數(shù)據(jù)流程時已經(jīng)詳細介紹過，這里不再贅述(《深入理解Kafka服務(wù)端之日志對象的讀寫數(shù)據(jù)流程》)。?對于后面的幾個步驟，由于發(fā)生在不同的節(jié)點上，并沒有按照這個順序進行分析，而是分成了
- Follower副本的相關(guān)操作：即 第三步、第八步、第九步
- Leader副本的相關(guān)操作：即 第四步、第五步、第六步、第七步
- 抽象類：AbstractFetcherThread
- 實現(xiàn)類：ReplicaFetcherThread
其中，type 的用法是：給指定的類起一個別名，如：
type FetchData = FetchResponse.PartitionData[Records]后面就可以用 FetchData 來表示 FetchResponse.PartitionData[Records] 類；EpochData 同理。
????FetchResponse.PartitionData：FetchResponse是封裝的FETCH請求的響應(yīng)類，PartitionData是一個嵌套類，表示響應(yīng)中單個分區(qū)的拉取信息，包括對應(yīng)Leader副本的高水位，分區(qū)日志的起始偏移量，拉取到的消息集合等。
public static final class PartitionData<T extends BaseRecords> { public final Errors error;//錯誤碼 public final long highWatermark;//從Leader返回的分區(qū)的高水位值 public final long lastStableOffset;// 最新LSO值 public final long logStartOffset;//日志起始偏移量 public final Optional preferredReadReplica;// 期望的Read Replica；KAFKA 2.4之后支持部分Follower副本可以對外提供讀服務(wù) public final List abortedTransactions;// 該分區(qū)對應(yīng)的已終止事務(wù)列表 public final T records;//消息集合}OffsetsForLeaderEpochRequest.PartitionData：里面包含了Follower副本在本地保存的leader epoch 和從Leader副本獲取到的leader epochpublic static class PartitionData { public final Optional currentLeaderEpoch; public final int leaderEpoch;}分區(qū)讀取的狀態(tài)：????PartitionFetchState：樣例類，用來表征分區(qū)的讀取狀態(tài)。包含已拉取的偏移量，當前l(fā)eader的epoch，副本讀取狀態(tài)等
case class PartitionFetchState(fetchOffset: Long,//已拉取的偏移量 currentLeaderEpoch: Int,//當前epoch delay: DelayedItem, state: ReplicaState//副本讀取狀態(tài) ) { //表征分區(qū)的讀取狀態(tài) //1.可拉取，表明副本獲取線程當前能夠讀取數(shù)據(jù)。判斷條件是：副本處于Fetching且未被推遲執(zhí)行 def isReadyForFetch: Boolean = state == Fetching && !isDelayed //2.截斷中，表明分區(qū)副本正在執(zhí)行截斷操作(比如該副本剛剛成為 Follower 副本)。判斷條件是：副本處于Truncating且未被推遲執(zhí)行 def isTruncating: Boolean = state == Truncating && !isDelayed //3.被推遲，表明副本獲取線程獲取數(shù)據(jù)時出現(xiàn)錯誤，需要等待一段時間后重試。判斷條件是：存在未過期的延遲任務(wù) def isDelayed: Boolean = delay.getDelay(TimeUnit.MILLISECONDS) > 0}分區(qū)讀取狀態(tài)分為三種：- isReadyForFetch：可拉取，表明副本獲取線程當前能夠讀取數(shù)據(jù)。判斷條件是：副本處于Fetching且未被推遲執(zhí)行
- isTruncating：截斷中，表明分區(qū)副本正在執(zhí)行截斷操作(比如該副本剛剛成為 Follower 副本)。判斷條件是：副本處于Truncating且未被推遲執(zhí)行
- isDelayed：被推遲，表明副本獲取線程獲取數(shù)據(jù)時出現(xiàn)錯誤，需要等待一段時間后重試。判斷條件是：存在未過期的延遲任務(wù)
副本讀取的狀態(tài)
????ReplicaState：特質(zhì)，用來表征副本讀取狀態(tài)。
sealed trait ReplicaState//截斷中case object Truncating extends ReplicaState//拉取中case object Fetching extends ReplicaState副本讀取狀態(tài)分為兩種：- Truncating：截斷中
- Fetching：拉取中
- buildFetch：封裝拉取數(shù)據(jù)的請求
- truncate：進行日志截斷
- processPartitionData：處理返回的響應(yīng)
- doWork：將上面定義的三個方法串聯(lián)起來，形成一個閉環(huán)，并不斷地重復(fù)執(zhí)行。從而實現(xiàn)從Leader副本所在的節(jié)點同步消息
buildFetch() 方法：為指定分區(qū)集合構(gòu)建對應(yīng)的FetchRequest.Builder 對象，而該對象是構(gòu)建 FetchRequest 的核心組件。
這個方法中有一個重要的操作：
封裝拉取請求時，攜帶了Follower副本的 LogStartOffset 和 LEO 值(對應(yīng)同步數(shù)據(jù)的第三步)
truncate() 方法：用于將指定分區(qū)的日志截斷到指定的偏移量
override def truncate(tp: TopicPartition, offsetTruncationState: OffsetTruncationState): Unit = { //根據(jù)分區(qū)獲取本地副本 val replica = replicaMgr.localReplicaOrException(tp) val partition = replicaMgr.getPartition(tp).get //調(diào)用Partition.truecateTo方法進行日志截斷 // offsetTruncationState.offset：要截斷到的偏移量 partition.truncateTo(offsetTruncationState.offset, isFuture = false) if (offsetTruncationState.offset < replica.highWatermark.messageOffset) warn(s"Truncating $tp to offset ${offsetTruncationState.offset} below high watermark " + s"${replica.highWatermark.messageOffset}") if (offsetTruncationState.truncationCompleted) replicaMgr.replicaAlterLogDirsManager.markPartitionsForTruncation(brokerConfig.brokerId, tp, offsetTruncationState.offset)}????這個方法內(nèi)部依次調(diào)用了：Partition.truncateTo -> LogManager.truncateTo -> Log.truncateTo -> LogSegment.truncateTo 進行日志截斷操作
processPartitionData方法：用于處理指定分區(qū)從Leader副本所在節(jié)點返回的響應(yīng)，將獲取的消息寫入本地存儲，并返回寫入消息的元數(shù)據(jù)
這里有兩個個重要的操作：
寫入消息，更新 Follower 副本的 LEO(對應(yīng)同步數(shù)據(jù)的第八步)
更新 Follower 副本本地的 HW 值(對應(yīng)同步數(shù)據(jù)的第九步)
這個方法很簡單，只在內(nèi)部調(diào)用了兩個方法：
maybeTruncate()：嘗試進行日志截斷
private def maybeTruncate(): Unit = { // 將所有處于截斷中狀態(tài)的分區(qū)依據(jù)有無Leader Epoch值進行分組 val (partitionsWithEpochs, partitionsWithoutEpochs) = fetchTruncatingPartitions() // 對于有Leader Epoch值的分區(qū)，將日志截斷到Leader Epoch值對應(yīng)的位移值處 if (partitionsWithEpochs.nonEmpty) { truncateToEpochEndOffsets(partitionsWithEpochs) } // 對于沒有Leader Epoch值的分區(qū)，將日志截斷到高水位值處 if (partitionsWithoutEpochs.nonEmpty) { truncateToHighWatermark(partitionsWithoutEpochs) }}這里先看對于沒有Leader Epoch的分區(qū)，將日志截斷到高水位處：
private[server] def truncateToHighWatermark(partitions: Set[TopicPartition]): Unit = inLock(partitionMapLock) { val fetchOffsets = mutable.HashMap.empty[TopicPartition, OffsetTruncationState] // 遍歷每個要執(zhí)行截斷操作的分區(qū)對象 for (tp // 獲取分區(qū)的分區(qū)讀取狀態(tài) val partitionState = partitionStates.stateValue(tp) if (partitionState != null) { // 取出高水位值。 val highWatermark = partitionState.fetchOffset //封裝截斷狀態(tài) val truncationState = OffsetTruncationState(highWatermark, truncationCompleted = true) info(s"Truncating partition $tp to local high watermark $highWatermark") // 執(zhí)行截斷到高水位值 if (doTruncate(tp, truncationState)) //保存分區(qū)和對應(yīng)的截取狀態(tài) fetchOffsets.put(tp, truncationState) } } // 更新這組分區(qū)的分區(qū)讀取狀態(tài) updateFetchOffsetAndMaybeMarkTruncationComplete(fetchOffsets)}其中 doTruncate(tp, truncationState) 方法內(nèi)部就調(diào)用了實現(xiàn)類：ReplicaFetcherThread.truncate() 方法maybeFetch()：嘗試從Leader副本拉取數(shù)據(jù)private def maybeFetch(): Unit = { //獲取分區(qū)狀態(tài)集合和對應(yīng)的拉取請求的集合 val (fetchStates, fetchRequestOpt) = inLock(partitionMapLock) { //獲取要拉取消息的分區(qū)和分區(qū)對應(yīng)狀態(tài)的集合 val fetchStates = partitionStates.partitionStateMap.asScala // TODO 第一步：為集合中的分區(qū)構(gòu)造FetchRequest.builder對象，這里的返回結(jié)果有兩個對象： //fetchRequestOpt：要讀取的分區(qū)核心信息 + FetchRequest.Builder 對象。 // 而這里的核心信息，就是指要讀取哪個分區(qū)，從哪個位置開始讀，最多讀多少字節(jié)，等等。 //partitionsWithError：一組出錯的分區(qū) val ResultWithPartitions(fetchRequestOpt, partitionsWithError) = buildFetch(fetchStates) //TODO 第二步：處理出錯的分區(qū)，處理方式主要是將這個分區(qū)加入到有序Map末尾，等待后續(xù)重試 handlePartitionsWithErrors(partitionsWithError, "maybeFetch") // 如果當前沒有可讀取的分區(qū)，則等待fetchBackOffMs時間等候后續(xù)重試 if (fetchRequestOpt.isEmpty) { trace(s"There are no active partitions. Back off for $fetchBackOffMs ms before sending a fetch request") partitionMapCond.await(fetchBackOffMs, TimeUnit.MILLISECONDS) } (fetchStates, fetchRequestOpt) } //TODO 第三步：遍歷FETCH請求，發(fā)送FETCH請求給Leader副本，并處理Response fetchRequestOpt.foreach { fetchRequest => processFetchRequest(fetchStates, fetchRequest) }}這個方法可以劃分為關(guān)鍵的三個步驟：a：為集合中的分區(qū)構(gòu)造FetchRequest.builder對象val ResultWithPartitions(fetchRequestOpt, partitionsWithError) = buildFetch(fetchStates)這里調(diào)用了實現(xiàn)類：ReplicaFetcherThread.buildFetch() 方法，返回結(jié)果有兩個對象：fetchRequestOpt：要讀取的分區(qū)核心信息 + FetchRequest.Builder 對象。而這里的核心信息，就是指要讀取哪個分區(qū)，從哪個位置開始讀，最多讀多少字節(jié)，等等。partitionsWithError：一組出錯的分區(qū)b：處理出錯的分區(qū)。處理方式主要是將這個分區(qū)加入到有序Map末尾，等待后續(xù)重試
handlePartitionsWithErrors(partitionsWithError, "maybeFetch")這個方法最后調(diào)用了PartitionStates.updateAndMoveToEnd() 方法，其作用就是把給定的分區(qū)從數(shù)據(jù)結(jié)構(gòu)的頭部移除，然后放到尾部，從而達到輪詢的目的//將給定的分區(qū)從map頭部移除，然后再加到尾部，以達到輪詢的目的//這里的LinkedHashMap對于插入元素是有順序的，加入插入順序是abcde，先讀取了a，// 為了保證公平性，會將a從集合中先移除，然后放到尾部，那么下次就從b開始讀public void updateAndMoveToEnd(TopicPartition topicPartition, S state) { map.remove(topicPartition); map.put(topicPartition, state); updateSize();}c：遍歷并發(fā)送FETCH請求給Leader副本，然后處理Response
fetchRequestOpt.foreach { fetchRequest => processFetchRequest(fetchStates, fetchRequest)}private def processFetchRequest(fetchStates: Map[TopicPartition, PartitionFetchState], fetchRequest: FetchRequest.Builder): Unit = { //定義出錯分區(qū)的集合 val partitionsWithError = mutable.Set[TopicPartition]() //定義接收響應(yīng)數(shù)據(jù)的集合 var responseData: Seq[(TopicPartition, FetchData)] = Seq.empty try { trace(s"Sending fetch request $fetchRequest") //給Leader發(fā)送FETCH請求，獲取響應(yīng)數(shù)據(jù) responseData = fetchFromLeader(fetchRequest) } catch { case t: Throwable => if (isRunning) { warn(s"Error in response for fetch request $fetchRequest", t) inLock(partitionMapLock) { partitionsWithError ++= partitionStates.partitionSet.asScala partitionMapCond.await(fetchBackOffMs, TimeUnit.MILLISECONDS) } } } //更新請求發(fā)送速率指標 fetcherStats.requestRate.mark() //如果接收到了響應(yīng) if (responseData.nonEmpty) { inLock(partitionMapLock) { //遍歷響應(yīng)結(jié)果中的分區(qū)和分區(qū)對應(yīng)的數(shù)據(jù) responseData.foreach { case (topicPartition, partitionData) => Option(partitionStates.stateValue(topicPartition)).foreach { currentFetchState => //獲取分區(qū)對應(yīng)的拉取狀態(tài) val fetchState = fetchStates(topicPartition) // 處理Response的條件： // 1. 獲取的消息集合的起始偏移量和之前已保存的下一條待寫入偏移量相等 // 2. 當前分區(qū)處于可獲取狀態(tài) if (fetchState.fetchOffset == currentFetchState.fetchOffset && currentFetchState.isReadyForFetch) { //獲取請求中攜帶的Follower副本保存的 leader epoch 值 val requestEpoch = if (fetchState.currentLeaderEpoch >= 0) Some(fetchState.currentLeaderEpoch) else None partitionData.error match { // 如果沒有錯誤 case Errors.NONE => try { // 交由子類完成Response的處理 val logAppendInfoOpt = processPartitionData(topicPartition, currentFetchState.fetchOffset, partitionData) logAppendInfoOpt.foreach { logAppendInfo => val validBytes = logAppendInfo.validBytes val nextOffset = if (validBytes > 0) logAppendInfo.lastOffset + 1 else currentFetchState.fetchOffset fetcherLagStats.getAndMaybePut(topicPartition).lag = Math.max(0L, partitionData.highWatermark - nextOffset) if (validBytes > 0 && partitionStates.contains(topicPartition)) { val newFetchState = PartitionFetchState(nextOffset, fetchState.currentLeaderEpoch, state = Fetching) // 將該分區(qū)放置在有序Map讀取順序的末尾，保證公平性 partitionStates.updateAndMoveToEnd(topicPartition, newFetchState) fetcherStats.byteRate.mark(validBytes) } } } catch { case ime: CorruptRecordException => error(s"Found invalid messages during fetch for partition $topicPartition " + s"offset ${currentFetchState.fetchOffset}", ime) partitionsWithError += topicPartition case e: KafkaStorageException => error(s"Error while processing data for partition $topicPartition " + s"at offset ${currentFetchState.fetchOffset}", e) markPartitionFailed(topicPartition) case t: Throwable => error(s"Unexpected error occurred while processing data for partition $topicPartition " + s"at offset ${currentFetchState.fetchOffset}", t) markPartitionFailed(topicPartition) } // 如果讀取位移值越界，通常是因為Leader發(fā)生變更 case Errors.OFFSET_OUT_OF_RANGE => //調(diào)整越界，主要辦法是做截斷 if (handleOutOfRangeError(topicPartition, currentFetchState, requestEpoch)) //如果依然不能成功，將該分區(qū)添加到出錯分區(qū)集合 partitionsWithError += topicPartition //如果Follower本地保存的Leader Epoch值比Leader所在Broker上的Epoch值要新 case Errors.UNKNOWN_LEADER_EPOCH => debug(s"Remote broker has a smaller leader epoch for partition $topicPartition than " + s"this replica's current leader epoch of ${fetchState.currentLeaderEpoch}.") // 加入到出錯分區(qū)集合 partitionsWithError += topicPartition // 如果Follower本地保存的Leader Epoch值比Leader所在Broker上的Epoch值要舊 case Errors.FENCED_LEADER_EPOCH => //將該分區(qū)標記為失效，從分區(qū)拉取狀態(tài)集合中移除，并加入到失效分區(qū)集合 if (onPartitionFenced(topicPartition, requestEpoch)) partitionsWithError += topicPartition // 如果Leader發(fā)生變更 case Errors.NOT_LEADER_FOR_PARTITION => debug(s"Remote broker is not the leader for partition $topicPartition, which could indicate " + "that the partition is being moved") // 加入到出錯分區(qū)列表 partitionsWithError += topicPartition case _ => error(s"Error for partition $topicPartition at offset ${currentFetchState.fetchOffset}", partitionData.error.exception) // 加入到出錯分區(qū)集合 partitionsWithError += topicPartition } } } } } } // 處理出錯分區(qū)集合，主要就是將該分區(qū)放到map數(shù)據(jù)結(jié)構(gòu)的末尾 if (partitionsWithError.nonEmpty) { handlePartitionsWithErrors(partitionsWithError, "processFetchRequest") } }Leader 副本如何處理拉取數(shù)據(jù)的請求：前面提到過，發(fā)送給服務(wù)端的各種請求都是由KafkaApis類處理的，處理FETCH請求的方法是：handleFetchRequest()內(nèi)部調(diào)用了ReplicaManager.fetchMessages() 方法：
def handleFetchRequest(request: RequestChannel.Request) { ... //TODO 這里是處理Follower Replica 拉取消息請求的具體方法 replicaManager.fetchMessages( fetchRequest.maxWait.toLong, fetchRequest.replicaId, fetchRequest.minBytes, fetchRequest.maxBytes, versionId <= 2, interesting, replicationQuota(fetchRequest), processResponseCallback, fetchRequest.isolationLevel) ...}fetchMessages() 方法：
def fetchMessages(timeout: Long, replicaId: Int, fetchMinBytes: Int, fetchMaxBytes: Int, hardMaxBytesLimit: Boolean, fetchInfos: Seq[(TopicPartition, PartitionData)], quota: ReplicaQuota = UnboundedQuota, responseCallback: Seq[(TopicPartition, FetchPartitionData)] => Unit, isolationLevel: IsolationLevel) { val isFromFollower = Request.isValidBrokerId(replicaId) val fetchOnlyFromLeader = replicaId != Request.DebuggingConsumerId && replicaId != Request.FutureLocalReplicaId val fetchIsolation = if (isFromFollower || replicaId == Request.FutureLocalReplicaId) FetchLogEnd else if (isolationLevel == IsolationLevel.READ_COMMITTED) FetchTxnCommitted else FetchHighWatermark //從本地磁盤讀取數(shù)據(jù) def readFromLog(): Seq[(TopicPartition, LogReadResult)] = { val result = readFromLocalLog( replicaId = replicaId, fetchOnlyFromLeader = fetchOnlyFromLeader, fetchIsolation = fetchIsolation, fetchMaxBytes = fetchMaxBytes, hardMaxBytesLimit = hardMaxBytesLimit, readPartitionInfo = fetchInfos, quota = quota) if (isFromFollower) updateFollowerLogReadResults(replicaId, result) else result } //獲取讀取結(jié)果 val logReadResults = readFromLog() var bytesReadable: Long = 0 var errorReadingData = false val logReadResultMap = new mutable.HashMap[TopicPartition, LogReadResult] logReadResults.foreach { case (topicPartition, logReadResult) => if (logReadResult.error != Errors.NONE) errorReadingData = true bytesReadable = bytesReadable + logReadResult.info.records.sizeInBytes logReadResultMap.put(topicPartition, logReadResult) } if (timeout <= 0 || fetchInfos.isEmpty || bytesReadable >= fetchMinBytes || errorReadingData) { val fetchPartitionData = logReadResults.map { case (tp, result) => tp -> FetchPartitionData(result.error, result.highWatermark, result.leaderLogStartOffset, result.info.records, result.lastStableOffset, result.info.abortedTransactions) } responseCallback(fetchPartitionData) } else { val fetchPartitionStatus = new mutable.ArrayBuffer[(TopicPartition, FetchPartitionStatus)] fetchInfos.foreach { case (topicPartition, partitionData) => logReadResultMap.get(topicPartition).foreach(logReadResult => { val logOffsetMetadata = logReadResult.info.fetchOffsetMetadata fetchPartitionStatus += (topicPartition -> FetchPartitionStatus(logOffsetMetadata, partitionData)) }) } val fetchMetadata = FetchMetadata(fetchMinBytes, fetchMaxBytes, hardMaxBytesLimit, fetchOnlyFromLeader, fetchIsolation, isFromFollower, replicaId, fetchPartitionStatus) val delayedFetch = new DelayedFetch(timeout, fetchMetadata, this, quota, responseCallback) val delayedFetchKeys = fetchPartitionStatus.map { case (tp, _) => new TopicPartitionOperationKey(tp) } delayedFetchPurgatory.tryCompleteElseWatch(delayedFetch, delayedFetchKeys) } }該方法內(nèi)部定義了一個readFromLog()方法，其作用有兩個：????a. 調(diào)用readFromLocalLog() 讀取 Leader 副本的本地日志????b. 調(diào)用 updateFollowerLogReadResults() 更新Leader副本的HW、Leader副本保存的對應(yīng)Follower副本的LEO，以及嘗試調(diào)整ISR列表等readFromLocalLog() 方法和內(nèi)部定義的 read() 方法如下，用于從Leader副本的日志文件讀取數(shù)據(jù)：def readFromLocalLog(replicaId: Int, fetchOnlyFromLeader: Boolean, fetchIsolation: FetchIsolation, fetchMaxBytes: Int, hardMaxBytesLimit: Boolean, readPartitionInfo: Seq[(TopicPartition, PartitionData)], quota: ReplicaQuota): Seq[(TopicPartition, LogReadResult)] = { def read(tp: TopicPartition, fetchInfo: PartitionData, limitBytes: Int, minOneMessage: Boolean): LogReadResult = { //讀取的起始偏移量 val offset = fetchInfo.fetchOffset //讀取的大小 val partitionFetchSize = fetchInfo.maxBytes //follower Replica 的LogStartOffset val followerLogStartOffset = fetchInfo.logStartOffset brokerTopicStats.topicStats(tp.topic).totalFetchRequestRate.mark() brokerTopicStats.allTopicsStats.totalFetchRequestRate.mark() val adjustedMaxBytes = math.min(fetchInfo.maxBytes, limitBytes) try { trace(s"Fetching log segment for partition $tp, offset $offset, partition fetch size $partitionFetchSize, " + s"remaining response limit $limitBytes" + (if (minOneMessage) s", ignoring response/partition size limits" else "")) val partition = getPartitionOrException(tp, expectLeader = fetchOnlyFromLeader) val fetchTimeMs = time.milliseconds //讀取數(shù)據(jù)，獲取讀取結(jié)果，里面包含了讀取到的消息，LEO，HW，LogStartOffset等信息 val readInfo = partition.readRecords( //讀取的起始偏移量 fetchOffset = fetchInfo.fetchOffset, //Follower副本保存的Leader epoch currentLeaderEpoch = fetchInfo.currentLeaderEpoch, maxBytes = adjustedMaxBytes, fetchIsolation = fetchIsolation, fetchOnlyFromLeader = fetchOnlyFromLeader, minOneMessage = minOneMessage) //獲取讀到的數(shù)據(jù) val fetchDataInfo = if (shouldLeaderThrottle(quota, tp, replicaId)) { //如果分區(qū)被限流了，那么返回一個空集合 FetchDataInfo(readInfo.fetchedData.fetchOffsetMetadata, MemoryRecords.EMPTY) } else if (!hardMaxBytesLimit && readInfo.fetchedData.firstEntryIncomplete) { //如果返回的消息集合不完整，也返回一個空集合 FetchDataInfo(readInfo.fetchedData.fetchOffsetMetadata, MemoryRecords.EMPTY) } else { //正常返回 readInfo.fetchedData } //根據(jù)獲取到的數(shù)據(jù)封裝返回結(jié)果 LogReadResult(info = fetchDataInfo, highWatermark = readInfo.highWatermark,//Leader的HW leaderLogStartOffset = readInfo.logStartOffset,//Leader的LogStartOffset leaderLogEndOffset = readInfo.logEndOffset,//Leader的LEO followerLogStartOffset = followerLogStartOffset,//Follower的LogStartOffset fetchTimeMs = fetchTimeMs, readSize = adjustedMaxBytes, lastStableOffset = Some(readInfo.lastStableOffset), exception = None//異常信息 ) } catch { case e@ (_: UnknownTopicOrPartitionException | _: NotLeaderForPartitionException | _: UnknownLeaderEpochException | _: FencedLeaderEpochException | _: ReplicaNotAvailableException | _: KafkaStorageException | _: OffsetOutOfRangeException) => LogReadResult(info = FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY), highWatermark = -1L, leaderLogStartOffset = -1L, leaderLogEndOffset = -1L, followerLogStartOffset = -1L, fetchTimeMs = -1L, readSize = 0, lastStableOffset = None, exception = Some(e)) case e: Throwable => brokerTopicStats.topicStats(tp.topic).failedFetchRequestRate.mark() brokerTopicStats.allTopicsStats.failedFetchRequestRate.mark() val fetchSource = Request.describeReplicaId(replicaId) error(s"Error processing fetch with max size $adjustedMaxBytes from $fetchSource " + s"on partition $tp: $fetchInfo", e) LogReadResult(info = FetchDataInfo(LogOffsetMetadata.UnknownOffsetMetadata, MemoryRecords.EMPTY), highWatermark = -1L, leaderLogStartOffset = -1L, leaderLogEndOffset = -1L, followerLogStartOffset = -1L, fetchTimeMs = -1L, readSize = 0, lastStableOffset = None, exception = Some(e)) } } //讀取的最大字節(jié) var limitBytes = fetchMaxBytes //封裝結(jié)果對象 val result = new mutable.ArrayBuffer[(TopicPartition, LogReadResult)] //是否至少返回一條消息 var minOneMessage = !hardMaxBytesLimit //遍歷分區(qū)進行讀取 readPartitionInfo.foreach { case (tp, fetchInfo) => //獲取讀取的結(jié)果 val readResult = read(tp, fetchInfo, limitBytes, minOneMessage) //獲取每個分區(qū)讀取的字節(jié)數(shù) val recordBatchSize = readResult.info.records.sizeInBytes if (recordBatchSize > 0) minOneMessage = false //更新還可以讀取的字節(jié)數(shù) limitBytes = math.max(0, limitBytes - recordBatchSize) //將分區(qū)的讀取結(jié)果保存到結(jié)果集合中 result += (tp -> readResult) } //返回結(jié)果集 result }其中，read() 方法中通過調(diào)用Partition. readRecords() 方法，就獲取了 Leader 副本的高水位值：//獲取Leader Replica的高水位val initialHighWatermark = localReplica.highWatermark.messageOffset從這里可以看出，每個分區(qū)的讀取結(jié)果中，都包含了 Leader 副本的 LEO、HW、LogStartOffset，以及 Follower 副本的LogStartOffset等信息。updateFollowerLogReadResults() 方法如下：private def updateFollowerLogReadResults(replicaId: Int, readResults: Seq[(TopicPartition, LogReadResult)]): Seq[(TopicPartition, LogReadResult)] = { debug(s"Recording follower broker $replicaId log end offsets: $readResults") readResults.map { case (topicPartition, readResult) => var updatedReadResult = readResult nonOfflinePartition(topicPartition) match { //如果找到了對應(yīng)的分區(qū) case Some(partition) => //根據(jù)副本id獲取Partition對象中保存的副本對象 //Partition.allReplicasMap結(jié)構(gòu)中保存了當前分區(qū)的所有副本對象。其中，key是brokerid，value是對應(yīng)的Replica對象 partition.getReplica(replicaId) match { //如果獲取到了Replica對象 case Some(replica) => //TODO 更新leader保存的各個follower副本的LEO partition.updateReplicaLogReadResult(replica, readResult) case None => warn(s"Leader $localBrokerId failed to record follower $replicaId's position " + s"${readResult.info.fetchOffsetMetadata.messageOffset} since the replica is not recognized to be " + s"one of the assigned replicas ${partition.assignedReplicas.map(_.brokerId).mkString(",")} " + s"for partition $topicPartition. Empty records will be returned for this partition.") updatedReadResult = readResult.withEmptyFetchInfo } //如果對應(yīng)的分區(qū)沒有被創(chuàng)建 case None => warn(s"While recording the replica LEO, the partition $topicPartition hasn't been created.") } topicPartition -> updatedReadResult } }Partition.updateReplicaLogReadResult() 方法：
def updateReplicaLogReadResult(replica: Replica, logReadResult: LogReadResult): Boolean = { val replicaId = replica.brokerId val oldLeaderLW = if (replicaManager.delayedDeleteRecordsPurgatory.delayed > 0) lowWatermarkIfLeader else -1L //TODO 最終更新Leader副本保存的Follower副本的LEO的值 replica.updateLogReadResult(logReadResult) val newLeaderLW = if (replicaManager.delayedDeleteRecordsPurgatory.delayed > 0) lowWatermarkIfLeader else -1L val leaderLWIncremented = newLeaderLW > oldLeaderLW //TODO 嘗試更新ISR列表，在這個方法中會更新Leader副本對象的HW對象和分區(qū)對應(yīng)的Log對象的HW值 val leaderHWIncremented = maybeExpandIsr(replicaId, logReadResult) val result = leaderLWIncremented || leaderHWIncremented if (result) tryCompleteDelayedRequests() debug(s"Recorded replica $replicaId log end offset (LEO) position ${logReadResult.info.fetchOffsetMetadata.messageOffset}.") result }Replica.updateLogReadResult() 方法：用于更新Partition保存的Follower副本的LEO(對應(yīng)同步數(shù)據(jù)的第四步)def updateLogReadResult(logReadResult: LogReadResult) { if (logReadResult.info.fetchOffsetMetadata.messageOffset >= logReadResult.leaderLogEndOffset) _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, logReadResult.fetchTimeMs) else if (logReadResult.info.fetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset) _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, lastFetchTimeMs) //更新Follower副本的日志起始偏移量，即 _logStartOffset 變量 logStartOffset = logReadResult.followerLogStartOffset //更新Follower副本的LEO元數(shù)據(jù)對象，即 _logEndOffsetMetadata 變量 logEndOffsetMetadata = logReadResult.info.fetchOffsetMetadata //最后一次拉取時Leader副本的LEO lastFetchLeaderLogEndOffset = logReadResult.leaderLogEndOffset lastFetchTimeMs = logReadResult.fetchTimeMs}maybeExpandIsr() 方法：嘗試更新ISR列表(對應(yīng)同步數(shù)據(jù)的第五步)def maybeExpandIsr(replicaId: Int, logReadResult: LogReadResult): Boolean = { inWriteLock(leaderIsrUpdateLock) { // 檢查給定的副本對象是否需要添加到ISR列表 leaderReplicaIfLocal match { case Some(leaderReplica) => //獲取給定節(jié)點的Replica對象 val replica = getReplica(replicaId).get //獲取leader副本的HW值 val leaderHW = leaderReplica.highWatermark //獲取Follower副本的LEO val fetchOffset = logReadResult.info.fetchOffsetMetadata.messageOffset //判斷是否需要更新ISR列表的條件： //1.該節(jié)點不在ISR列表，且replica.logEndOffsetMetadata.offsetDiff(leaderHW) //2.給定Follower副本的LEO大于等于leader副本的HW //3.給定的Follower副本屬于該分區(qū) //4.leader epoch對應(yīng)的起始偏移量存在且小于Follower副本的LEO //滿足這4個條件說明這個Follower副本已經(jīng)和leader副本保持同步了，把這個Follower副本加入到ISR列表中 if (!inSyncReplicas.contains(replica) && assignedReplicas.map(_.brokerId).contains(replicaId) && replica.logEndOffsetMetadata.offsetDiff(leaderHW) >= 0 && leaderEpochStartOffsetOpt.exists(fetchOffset >= _)) { //將該副本加入集合 val newInSyncReplicas = inSyncReplicas + replica info(s"Expanding ISR from ${inSyncReplicas.map(_.brokerId).mkString(",")} " + s"to ${newInSyncReplicas.map(_.brokerId).mkString(",")}") // update ISR in ZK and cache //更新ISR列表 updateIsr(newInSyncReplicas) replicaManager.isrExpandRate.mark() } // check if the HW of the partition can now be incremented // since the replica may already be in the ISR and its LEO has just incremented //TODO 嘗試更新leader副本的HW對象及分區(qū)對應(yīng)的Log對象的HW值 maybeIncrementLeaderHW(leaderReplica, logReadResult.fetchTimeMs) case None => false // nothing to do if no longer leader } } }maybeIncrementLeaderHW() 方法：嘗試更新 leader 副本的 HW 對象及分區(qū)對應(yīng)的Log 對象的 HW 值(對應(yīng)同步數(shù)據(jù)的第六步)
private def maybeIncrementLeaderHW(leaderReplica: Replica, curTime: Long = time.milliseconds): Boolean = { val allLogEndOffsets = assignedReplicas.filter { replica => curTime - replica.lastCaughtUpTimeMs <= replicaLagTimeMaxMs || inSyncReplicas.contains(replica) }.map(_.logEndOffsetMetadata) //取ISR列表中副本的最小的LEO作為新的HW val newHighWatermark = allLogEndOffsets.min(new LogOffsetMetadata.OffsetOrdering) //獲取舊的HW val oldHighWatermark = leaderReplica.highWatermark //如果新的HW值大于舊的HW值，就更新 if (oldHighWatermark.messageOffset < newHighWatermark.messageOffset || (oldHighWatermark.messageOffset == newHighWatermark.messageOffset && oldHighWatermark.onOlderSegment(newHighWatermark))) { //更新Replica的hightWatermark對象，以及對應(yīng)Log對象的高水位值 leaderReplica.highWatermark = newHighWatermark debug(s"High watermark updated to $newHighWatermark") true } else { def logEndOffsetString(r: Replica) = s"replica ${r.brokerId}: ${r.logEndOffsetMetadata}" debug(s"Skipping update high watermark since new hw $newHighWatermark is not larger than old hw $oldHighWatermark. " + s"All current LEOs are ${assignedReplicas.map(logEndOffsetString)}") false } }在前面分析Log日志對象的主要操作時，其中有一項是進行高水位操作的管理。在Log類中，操作高水位值的方法只有一個：onHighWatermarkIncrementeddef onHighWatermarkIncremented(highWatermark: Long): Unit = { lock synchronized { //更新高水位值??????replicaHighWatermark?=?Some(highWatermark)??????producerStateManager.onHighWatermarkUpdated(highWatermark) updateFirstUnstableOffset() } }這個方法就是將 Log 中的 replicaHightWatermark 變量修改為給定的值。那么什么時候會修改呢？查看調(diào)用該方法的地方：Replica.highWatermark_def highWatermark_=(newHighWatermark: LogOffsetMetadata) { //如果是本地副本 if (isLocal) { if (newHighWatermark.messageOffset < 0) throw new IllegalArgumentException("High watermark offset should be non-negative") //高水位的元數(shù)據(jù)對象 highWatermarkMetadata = newHighWatermark //更新Log對象保存的高水位值 log.foreach(_.onHighWatermarkIncremented(newHighWatermark.messageOffset)) trace(s"Setting high watermark for replica $brokerId partition $topicPartition to [$newHighWatermark]") } else { throw new KafkaException(s"Should not set high watermark on partition $topicPartition's non-local replica $brokerId") } }在嘗試更新Leader副本的高水位時會進行highWatermark_的調(diào)用：//更新Replica的hightWatermark對象，以及對應(yīng)Log對象的高水位值leaderReplica.highWatermark = newHighWatermark最后會將多個分區(qū)的讀取結(jié)果(包含Leader副本 HW)放到集合中，然后在合適的時機返回給Follower副本所在的節(jié)點(對應(yīng)同步數(shù)據(jù)第七步)def?fetchMessages(){ ... logReadResults.foreach { case (topicPartition, logReadResult) => //如果讀取發(fā)生錯誤 if (logReadResult.error != Errors.NONE) errorReadingData = true bytesReadable = bytesReadable + logReadResult.info.records.sizeInBytes //將讀取結(jié)果放入集合 logReadResultMap.put(topicPartition, logReadResult) }????...}????上面所說的合適的時機，分為 立即返回 和 延時返回，當滿足下面四個條件之一時，便立即返回，否則會進行延時處理：
拉取等待的時間到了
拉取請求中沒有拉取分區(qū)的信息
已經(jīng)拉取到了足夠多的數(shù)據(jù)
拉取過程中發(fā)生錯誤
Leader副本寫入數(shù)據(jù)，Follower副本進行同步的過程分為9個步驟：
關(guān)于 HW 和 LEO 的保存：
對于HW，Leader 副本和 Follower 副本只保存自身的
對于LEO，Follower 副本只保存自身的，但是 Leader 副本除了保存自身的外，還會保存所有 Follower 副本的 LEO 值
無論是Leader副本所在節(jié)點，還是Follower副本所在節(jié)點，分區(qū)對應(yīng)的Partition 對象都會保存所有的副本對象，但是只有本地副本對象有對應(yīng)的日志文件
總結(jié)
以上是生活随笔為你收集整理的取本地数据_深入理解Kafka服务端之Follower副本如何同步Leader副本的数据的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： MSSQL SQL简繁转换函数
- 下一篇： 《史蒂夫·乔布斯传》PDF下载