深入理解Kafka服务端之失效副本的判定及ISR列表的伸缩

def startup() {
    //由参数 replica.lag.time.max.ms  配置，默认为10秒
    scheduler.schedule("isr-expiration", maybeShrinkIsr _, period = config.replicaLagTimeMaxMs  2, unit = TimeUnit.MILLISECONDS)
    scheduler.schedule("isr-change-propagation", maybePropagateIsrChanges _, period = 2500L, unit = TimeUnit.MILLISECONDS)
    ...
}

private def maybeShrinkIsr(): Unit = {
    trace("Evaluating ISR list of partitions to see which replicas can be removed from the ISR")
    //遍历ReplicaManager管理的所有在线的Partition对象，检查是否需要缩小ISR列表
    allPartitions.keys.foreach { topicPartition =>
      nonOfflinePartition(topicPartition).foreach(_.maybeShrinkIsr(config.replicaLagTimeMaxMs))
    }
  }

def maybeShrinkIsr(replicaMaxLagTimeMs: Long) {
  val leaderHWIncremented = inWriteLock(leaderIsrUpdateLock) {
    leaderReplicaIfLocal match {
      //如果是本地Leader副本
      case Some(leaderReplica) =>
        //获取要被移除出去的Replica集合
        val outOfSyncReplicas = getOutOfSyncReplicas(leaderReplica, replicaMaxLagTimeMs)
        if (outOfSyncReplicas.nonEmpty) {
          //构建一个新的ISR列表，将要移除的Replica集合从原ISR列表移除
          val newInSyncReplicas = inSyncReplicas -- outOfSyncReplicas
          assert(newInSyncReplicas.nonEmpty)
          info("Shrinking ISR from %s to %s. Leader: (highWatermark: %d, endOffset: %d). Out of sync replicas: %s."
            .format(inSyncReplicas.map(_.brokerId).mkString(","),
              newInSyncReplicas.map(_.brokerId).mkString(","),
              leaderReplica.highWatermark.messageOffset,
              leaderReplica.logEndOffset,
              outOfSyncReplicas.map { replica =>
                s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset})"
              }.mkString(" ")
            )
          )
          //更新zk上和缓存中的ISR列表
          updateIsr(newInSyncReplicas)
          replicaManager.isrShrinkRate.mark()
          
          //尝试更新HW高水位值
          maybeIncrementLeaderHW(leaderReplica)
        } else {
          false
        }


      case None => false // 什么都不做
    }
  }
  //如果 HW 进行了修改
  if (leaderHWIncremented)
    //尝试完成所有延迟的请求
    tryCompleteDelayedRequests()
}

//获取失效的Replica集合
val outOfSyncReplicas = getOutOfSyncReplicas(leaderReplica, replicaMaxLagTimeMs)

def getOutOfSyncReplicas(leaderReplica: Replica, maxLagMs: Long): Set[Replica] = {
  //获取除Leader副本之外的其他副本集合
  val candidateReplicas = inSyncReplicas - leaderReplica


  //过滤失效的副本
  //失效副本判定条件：
  // 1.r.logEndOffset != leaderReplica.logEndOffset：该副本的LEO 不等于 Leader 副本的LEO
  // 2.(time.milliseconds - r.lastCaughtUpTimeMs) > maxLagMs  (当前时间 - Follower副本最后一次追上Leader副本的时间) > 最大的延迟时间
  //如果一个Replica长时间(10秒)没有发送请求到Leader 副本同步数据
  val laggingReplicas = candidateReplicas.filter(r =>
    r.logEndOffset != leaderReplica.logEndOffset && (time.milliseconds - r.lastCaughtUpTimeMs) > maxLagMs)
  if (laggingReplicas.nonEmpty)
    debug("Lagging replicas are %s".format(laggingReplicas.map(_.brokerId).mkString(",")))


  laggingReplicas
}

def updateLogReadResult(logReadResult: LogReadResult) {
  //更新最后一次追上Leader副本的时间 _lastCaughtUpTimeMs
  // 如果拉取结果的最大偏移量>=当前leader副本的LEO，也就是拉取到了小于LEO的所有数据，则更新_lastCaughtUpTimeMs为 max(_lastCaughtUpTimeMs,拉取到数据的时间)
  if (logReadResult.info.fetchOffsetMetadata.messageOffset >= logReadResult.leaderLogEndOffset)
    _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, logReadResult.fetchTimeMs)
  //如果拉取结果的最大偏移量>=上一次拉取时Leader副本的LEO，这表示Follower副本当前还没追上Leader副本
  // 这时更新_lastCaughtUpTimeMs为 max(_lastCaughtUpTimeMs,上一次拉取数据的时间）
  else if (logReadResult.info.fetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset)
    _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, lastFetchTimeMs)
  //更新Follower副本的日志起始偏移量，即 _logStartOffset 变量
  logStartOffset = logReadResult.followerLogStartOffset
  //更新Follower副本的LEO元数据对象，即 _logEndOffsetMetadata 变量
  logEndOffsetMetadata = logReadResult.info.fetchOffsetMetadata
  //更新最后一次拉取时Leader副本的LEO
  lastFetchLeaderLogEndOffset = logReadResult.leaderLogEndOffset
  //更新最后一次拉取数据的时间为从leader副本拉取到数据的时间
  lastFetchTimeMs = logReadResult.fetchTimeMs
}

if (outOfSyncReplicas.nonEmpty) {
    //构建一个新的ISR列表，是原始ISR列表剔除失效副本的结果
    val newInSyncReplicas = inSyncReplicas -- outOfSyncReplicas
    ...
    更新zk上和缓存中的ISR列表
    updateIsr(newInSyncReplicas)
    replicaManager.isrShrinkRate.mark()

private def updateIsr(newIsr: Set[Replica]) {
  //封装一个对象，包含 当前节点id，当前leaderepoch值，新的ISR列表中副本所在节点编号集合，zk版本号
  val newLeaderAndIsr = new LeaderAndIsr(localBrokerId, leaderEpoch, newIsr.map(_.brokerId).toList, zkVersion)
  //更新ZK上指定分区的节点数据，返回两个对象：
  //1.updateSucceeded：标识是否更新成功
  //2.新的版本号
  val (updateSucceeded, newVersion) = ReplicationUtils.updateLeaderAndIsr(zkClient, topicPartition, newLeaderAndIsr,
    controllerEpoch)
  //如果更新成功
  if (updateSucceeded) {
    //将该分区添加到isrChangeSet集合
    replicaManager.recordIsrChange(topicPartition)
    //更新ISR列表，注意这里只有在zk写成功后才会更改原来的ISR列表
    inSyncReplicas = newIsr
    //更新zk版本号
    zkVersion = newVersion
    trace("ISR updated to [%s] and zkVersion updated to [%d]".format(newIsr.mkString(","), zkVersion))
  } else {
    replicaManager.failedIsrUpdatesRate.mark()
    info("Cached zkVersion [%d] not equal to that in zookeeper, skip updating ISR".format(zkVersion))
  }
}

//更新指定分区在zk节点上的数据
def updateLeaderAndIsr(zkClient: KafkaZkClient, partition: TopicPartition, newLeaderAndIsr: LeaderAndIsr,
                       controllerEpoch: Int): (Boolean, Int) = {
  debug(s"Updated ISR for $partition to ${newLeaderAndIsr.isr.mkString(",")}")
  //该分区在zk上的节点路径，/brokers/topics/<topic>/partition/<partition>/state 节点
  val path = TopicPartitionStateZNode.path(partition)
  //zk节点对应的数据，包含controller_epoch,leader,version,leader_epoch,isr
  val newLeaderData = TopicPartitionStateZNode.encode(LeaderIsrAndControllerEpoch(newLeaderAndIsr, controllerEpoch))
  // use the epoch of the controller that made the leadership decision, instead of the current controller epoch
  val updatePersistentPath: (Boolean, Int) = zkClient.conditionalUpdatePath(path, newLeaderData,
    newLeaderAndIsr.zkVersion, Some(checkLeaderAndIsrZkData))
  //如果更新成功，返回(true,newVersion)
  updatePersistentPath
}

//尝试更新HW高水位值
maybeIncrementLeaderHW(leaderReplica)

//如果 HW 进行了修改
if (leaderHWIncremented)
  //尝试完成所有延迟的请求
  tryCompleteDelayedRequests()

def maybePropagateIsrChanges() {
  //获取当前时间
  val now = System.currentTimeMillis()
  isrChangeSet synchronized {
    //广播ISR变化的条件：
    //1.isrChangeSet不为空，即有分区更新了ISR列表
    //  2.1 距离上一次isrChangeSet更新的时间已经超过5秒( 上一次isrChangeSet更新的时间 + 5秒 < 当前时间)
    //  2.2 距离上一次写入ZK的时间已经超过60秒
    //IsrChangePropagationBlackOut 默认 5秒；IsrChangePropagationInterval 默认 60秒
    if (isrChangeSet.nonEmpty &&
      (lastIsrChangeMs.get() + ReplicaManager.IsrChangePropagationBlackOut < now ||
        lastIsrPropagationMs.get() + ReplicaManager.IsrChangePropagationInterval < now)) {
      //广播ISR列表的变化
      zkClient.propagateIsrChanges(isrChangeSet)
      //清空isrChangeSet集合
      isrChangeSet.clear()
      //更新写入ZK的时间
      lastIsrPropagationMs.set(now)
    }
  }
}

def propagateIsrChanges(isrChangeSet: collection.Set[TopicPartition]): Unit = {
  //在ZK的/isr_change_notification路径下创建一个"isr_change_"+ 单调递增数的顺序持久节点，并将isrChangeSet的信息保存到这个节点
  / 返回节点路径，如：/isr_change_notification/isr_change_0000000000
  val isrChangeNotificationPath: String = createSequentialPersistentPath(IsrChangeNotificationSequenceZNode.path(),
    IsrChangeNotificationSequenceZNode.encode(isrChangeSet))
  debug(s"Added $isrChangeNotificationPath for $isrChangeSet")
}