深入理解Kafka服务端之消费者组元数据的管理

case class MemberSummary(memberId: String,// 成员ID，由Kafka自动生成
                         groupInstanceId: Option[String],// Consumer端参数group.instance.id值
                         clientId: String,// client.id参数值
                         clientHost: String,// Consumer端程序主机名
                         metadata: Array[Byte],// 消费者组成员使用的分配策略
                         assignment: Array[Byte])// 成员订阅分区

private object MemberMetadata {
  // 提取分区分配策略集合
  def plainProtocolSet(supportedProtocols: List[(String, Array[Byte])]) = supportedProtocols.map(_._1).toSet
}

private[group] class MemberMetadata(var memberId: String,
                                    val groupId: String,
                                    val groupInstanceId: Option[String],
                                    val clientId: String,
                                    val clientHost: String,
                                    val rebalanceTimeoutMs: Int,// Rebalane操作超时时间
                                    val sessionTimeoutMs: Int,// 会话超时时间
                                    val protocolType: String,// 对消费者组而言，是"consumer"
                                    var supportedProtocols: List[(String, Array[Byte])]// 成员配置的多套分区分配策略
                                   ) {
  // 分区分配方案
  var assignment: Array[Byte] = Array.empty[Byte]
  //表示组成员是否正在等待加入组。
  var awaitingJoinCallback: JoinGroupResult => Unit = null
  //表示组成员是否正在等待 GroupCoordinator 发送分配方案。
  var awaitingSyncCallback: SyncGroupResult => Unit = null
  //表示组成员是否发起“退出组”的操作。
  var isLeaving: Boolean = false
  //表示是否是消费者组下的新成员。
  var isNew: Boolean = false
  val isStaticMember: Boolean = groupInstanceId.isDefined
  //用于标识心跳是否完成，当接收到心跳时为true，当超时未接到到心跳，设置为false
  var heartbeatSatisfied: Boolean = false
  ...
}

private[group] sealed trait GroupState


private[group] case object PreparingRebalance extends GroupState


...

/**
 * transition: some members have joined by the timeout => CompletingRebalance
 *             all members have left the group => Empty
 *             group is removed by partition emigration => Dead
 */

case class GroupOverview(groupId: String,//消费者组id，即：group.id 参数的值
                         protocolType: String//协议类型
                         )

case class GroupSummary(state: String,//消费者组状态
                        protocolType: String,//协议类型
                        protocol: String,//消费者组选定的分区分配策略
                        members: List[MemberSummary]//消费者组成员元数据
                       )

private object GroupMetadata {
  //用于存储各种状态的合法前置状态
  private val validPreviousStates: Map[GroupState, Set[GroupState]] =
    Map(Dead -> Set(Stable, PreparingRebalance, CompletingRebalance, Empty, Dead),
      CompletingRebalance -> Set(PreparingRebalance),
      Stable -> Set(CompletingRebalance),
      PreparingRebalance -> Set(Stable, CompletingRebalance, Empty),
      Empty -> Set(PreparingRebalance))
      ...
}

private[group] class GroupMetadata(val groupId: String, //消费者组id
                                   initialState: GroupState, //消费者组初始状态
                                   time: Time) extends Logging {
  type JoinCallback = JoinGroupResult => Unit


  private[group] val lock = new ReentrantLock
  // 组状态
  private var state: GroupState = initialState
  // 记录状态最近一次变更的时间戳
  var currentStateTimestamp: Option[Long] = Some(time.milliseconds())
  //协议类型
  var protocolType: Option[String] = None
  //记录rebalance的次数
  var generationId = 0
  // 记录消费者组的Leader成员，可能不存在
  private var leaderId: Option[String] = None
  //分区分配策略
  private var protocol: Option[String] = None
  // 成员元数据列表信息
  private val members = new mutable.HashMap[String, MemberMetadata]
  // Static membership mapping [key: group.instance.id, value: member.id]
  // 静态成员Id列表
  private val staticMembers = new mutable.HashMap[String, String]
  private val pendingMembers = new mutable.HashSet[String]
  private var numMembersAwaitingJoin = 0
  // 分区分配策略支持票数
  private val supportedProtocols = new mutable.HashMap[String, Integer]().withDefaultValue(0)
  // 保存消费者组订阅分区的提交位移值
  private val offsets = new mutable.HashMap[TopicPartition, CommitRecordMetadataAndOffset]
  private val pendingOffsetCommits = new mutable.HashMap[TopicPartition, OffsetAndMetadata]
  private val pendingTransactionalOffsetCommits = new mutable.HashMap[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]]()
  private var receivedTransactionalOffsetCommits = false
  private var receivedConsumerOffsetCommits = false


  var newMemberAdded: Boolean = false
  ...
}

//判断消费者组的当前状态是否为指定状态
def is(groupState: GroupState) = state == groupState


//判断消费者组的当前状态不是指定状态
def not(groupState: GroupState) = state != groupState


//获取消费者组当前状态
def currentState = state


//消费者组能否Rebalance的条件是当前状态是PreparingRebalance状态的合法前置状态
def canRebalance = GroupMetadata.validPreviousStates(PreparingRebalance).contains(state)


//设置/更新消费者组状态
def transitionTo(groupState: GroupState) {
  //验证目标状态的合法性
  assertValidTransition(groupState)
  //更新/设置消费者组状态
  state = groupState
  //设置更新状态的时间戳
  currentStateTimestamp = Some(time.milliseconds())
}

def add(member: MemberMetadata, callback: JoinCallback = null) {
    //如果members为空，说明添加的是第一个消费者
    if (members.isEmpty)
    // 就把该成员的protocolType设置为消费者组的protocolType
      this.protocolType = Some(member.protocolType)
    // 确保成员元数据中的groupId和组Id相同
    assert(groupId == member.groupId)
    // 确保成员元数据中的protoclType和组protocolType相同
    assert(this.protocolType.orNull == member.protocolType)
    // 确保该成员选定的分区分配策略与组选定的分区分配策略相匹配
    assert(supportsProtocols(member.protocolType, MemberMetadata.plainProtocolSet(member.supportedProtocols)))
    // 如果尚未选出Leader成员
    if (leaderId.isEmpty)
      // 把该成员设定为Leader成员
      leaderId = Some(member.memberId)
    // 将该成员添加进members
    members.put(member.memberId, member)
    // 更新分区分配策略支持票数
    member.supportedProtocols.foreach{ case (protocol, _) => supportedProtocols(protocol) += 1 }
    // 设置成员加入组后的回调逻辑
    member.awaitingJoinCallback = callback
    // 更新已加入组的成员数
    if (member.isAwaitingJoin)
      numMembersAwaitingJoin += 1
}

def remove(memberId: String) {
    //第一步：将成员从 members 变量中移除
    members.remove(memberId).foreach { member =>
      //第二步：更新分区分配策略支持票数
      member.supportedProtocols.foreach{ case (protocol, _) => supportedProtocols(protocol) -= 1 }
      //第三步：更新组中成员数量
      if (member.isAwaitingJoin)
        numMembersAwaitingJoin -= 1
    }
    //第四步：如果该成员是Leader，则重选Leader成员，策略就是选剩余成员列表中的第一个为Leader
    if (isLeader(memberId))
      leaderId = members.keys.headOption
}

//判断给定成员是否在组中
def has(memberId: String) = members.contains(memberId)
//根据成员ID获取成员
def get(memberId: String) = members(memberId)
//或者组成员的数量
def size = members.size

/ 保存消费者组订阅分区的提交位移值
private val offsets = new mutable.HashMap[TopicPartition, CommitRecordMetadataAndOffset]

case class CommitRecordMetadataAndOffset(appendedBatchOffset: Option[Long], offsetAndMetadata: OffsetAndMetadata) {
  def olderThan(that: CommitRecordMetadataAndOffset): Boolean = appendedBatchOffset.get < that.appendedBatchOffset.get
}

def initializeOffsets(offsets: collection.Map[TopicPartition, CommitRecordMetadataAndOffset],
                        pendingTxnOffsets: Map[Long, mutable.Map[TopicPartition, CommitRecordMetadataAndOffset]]) {
    //将给定的一组订阅分区提交位移值加到 offsets 中
    this.offsets ++= offsets
    //更新 pendingTransactionalOffsetCommits
    this.pendingTransactionalOffsetCommits ++= pendingTxnOffsets
}

def onOffsetCommitAppend(topicPartition: TopicPartition, offsetWithCommitRecordMetadata: CommitRecordMetadataAndOffset) {
    if (pendingOffsetCommits.contains(topicPartition)) {
      if (offsetWithCommitRecordMetadata.appendedBatchOffset.isEmpty)
        throw new IllegalStateException("Cannot complete offset commit write without providing the metadata of the record " +
          "in the log.")
      // offsets字段中没有该分区位移提交数据，或者
      // offsets字段中该分区对应的提交位移消息在位移主题中的位移值小于待写入的位移值
      if (!offsets.contains(topicPartition) || offsets(topicPartition).olderThan(offsetWithCommitRecordMetadata))
      // 将该分区对应的提交位移消息添加到offsets中
        offsets.put(topicPartition, offsetWithCommitRecordMetadata)
    }


    pendingOffsetCommits.get(topicPartition) match {
      case Some(stagedOffset) if offsetWithCommitRecordMetadata.offsetAndMetadata == stagedOffset =>
        pendingOffsetCommits.remove(topicPartition)
      case _ =>
        // The pendingOffsetCommits for this partition could be empty if the topic was deleted, in which case
        // its entries would be removed from the cache by the `removeOffsets` method.
    }
}

def removeExpiredOffsets(currentTimestamp: Long, offsetRetentionMs: Long): Map[TopicPartition, OffsetAndMetadata] = {
    //用于获取订阅分区过期的位移值
    def getExpiredOffsets(baseTimestamp: CommitRecordMetadataAndOffset => Long): Map[TopicPartition, OffsetAndMetadata] = {
      // 遍历offsets中的所有分区，过滤出同时满足以下2个条件的分区
      // 条件1：该主题分区已经完成位移提交
      // 条件2：该主题分区在位移主题中对应消息的存在时间超过了阈值
      offsets.filter {
        case (topicPartition, commitRecordMetadataAndOffset) =>
          !pendingOffsetCommits.contains(topicPartition) && {
            //获取位移消息中的时间戳
            commitRecordMetadataAndOffset.offsetAndMetadata.expireTimestamp match {
              //如果没有时间戳
              case None =>
                // current version with no per partition retention
                currentTimestamp - baseTimestamp(commitRecordMetadataAndOffset) >= offsetRetentionMs
              //如果获取到了时间戳
              case Some(expireTimestamp) =>


                //判断条件就是方法传入的currentTimestamp大于等于消息中的时间戳
                currentTimestamp >= expireTimestamp
            }
          }
      }.map {
        case (topicPartition, commitRecordOffsetAndMetadata) =>
          //将位移值对象返回
          (topicPartition, commitRecordOffsetAndMetadata.offsetAndMetadata)
      }.toMap
    }
    // 调用getExpiredOffsets方法获取主题分区的过期位移
    val expiredOffsets: Map[TopicPartition, OffsetAndMetadata] = protocolType match {
      case Some(_) if is(Empty) =>
        //传入更新为当前组状态的时间戳
        getExpiredOffsets(commitRecordMetadataAndOffset =>
          currentStateTimestamp.getOrElse(commitRecordMetadataAndOffset.offsetAndMetadata.commitTimestamp))
      //如果没有定义协议类型
      case None =>


        getExpiredOffsets(_.offsetAndMetadata.commitTimestamp)


      case _ =>
        Map()
    }


    if (expiredOffsets.nonEmpty)
      debug(s"Expired offsets from group '$groupId': ${expiredOffsets.keySet}")
    //将过期位移对应的主题分区从offsets中移除
    offsets --= expiredOffsets.keySet
    //返回主题分区对应的过期位移
    expiredOffsets
}

private def candidateProtocols = {
    //找出支持票数=总成员数的策略，返回它们的名称
    val numMembers = members.size
    supportedProtocols.filter(_._2 == numMembers).map(_._1).toSet
}

def selectProtocol: String = {
    //如果没有成员，抛出异常
    if (members.isEmpty)
      throw new IllegalStateException("Cannot select protocol for empty group")


    // select the protocol for this group which is supported by all members
    //获取所有成员都支持的分配策略
    val candidates = candidateProtocols


    // let each member vote for one of the protocols and choose the one with the most votes
    //让每个成员投票，票数最多的那个策略当选
    val votes: List[(String, Int)] = allMemberMetadata
      .map(_.vote(candidates))
      .groupBy(identity)
      .mapValues(_.size)
      .toList


    votes.maxBy(_._2)._1
}

def vote(candidates: Set[String]): String = {
    //找到当前成员配置的分区分配策略中，第一个在给定的分区分配策略中的
    supportedProtocols.find({ case (protocol, _) => candidates.contains(protocol)}) match {
      case Some((protocol, _)) => protocol
      case None =>
        throw new IllegalArgumentException("Member does not support any of the candidate protocols")
    }
}