1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private[group] class MemberMetadata(var memberId: String, // 成员ID
                                    val groupId: String, // 消费组ID
                                    val groupInstanceId: Option[String],
                                    val clientId: String,
                                    val clientHost: String,
                                    val rebalanceTimeoutMs: Int,
                                    val sessionTimeoutMs: Int,
                                    val protocolType: String,
                                    var supportedProtocols: List[(String, Array[Byte])]) {

  var assignment: Array[Byte] = Array.empty[Byte] // 消费者分区结果
  var awaitingJoinCallback: JoinGroupResult => Unit = null // JOIN_GROUP回调
  var awaitingSyncCallback: SyncGroupResult => Unit = null // SYNC_GROUP回调
  var isLeaving: Boolean = false
  var isNew: Boolean = false
  val isStaticMember: Boolean = groupInstanceId.isDefined
  var heartbeatSatisfied: Boolean = false
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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 // 组状态: Empty(初始)/Dead/Stable/PreparingRebalance/CompletingRebalance
  var currentStateTimestamp: Option[Long] = Some(time.milliseconds())
  var protocolType: Option[String] = None // 协议类型
  var protocolName: Option[String] = None // 协议名称
  var generationId = 0 // Generation号
  private var leaderId: Option[String] = None // 组Leader ID

  private val members = new mutable.HashMap[String, MemberMetadata] // 组成员
  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
  private var subscribedTopics: Option[Set[String]] = None // 已订阅的topic集合
  var newMemberAdded: Boolean = false
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 返回响应时的回调
def sendResponseCallback(joinResult: JoinGroupResult): Unit = {
  // 响应创建函数
  def createResponse(requestThrottleMs: Int): AbstractResponse = {
    val protocolName = if (request.context.apiVersion() >= 7)
      joinResult.protocolName.orNull
    else
      joinResult.protocolName.getOrElse(GroupCoordinator.NoProtocol)

    val responseBody = new JoinGroupResponse(
      new JoinGroupResponseData()
        .setThrottleTimeMs(requestThrottleMs)
        .setErrorCode(joinResult.error.code) // 错误码
        .setGenerationId(joinResult.generationId) // generation号
        .setProtocolType(joinResult.protocolType.orNull) // 协议类型
        .setProtocolName(protocolName) // 协议名称
        .setLeader(joinResult.leaderId) // Leader ID（主消费者）
        .setMemberId(joinResult.memberId) // 当前消费者ID
        .setMembers(joinResult.members.asJava) // 已知消费组的成员
    )
    responseBody
  }
  // 返回响应
  sendResponseMaybeThrottle(request, createResponse)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def onCompleteJoin(group: GroupMetadata): Unit = {
  group.inLock {
    // ... 

    if /* ... */ {
      // ...
    } else {
      group.initNextGeneration() // 增加generation
      if (group.is(Empty)) {
        // ...
      } else {
        // 一次性向每个消费者返回响应
        for (member <- group.allMemberMetadata) {
          val joinResult = JoinGroupResult(
            members = if (group.isLeader(member.memberId)) {
              group.currentMemberMetadata
            } else {
              List.empty
            },
            memberId = member.memberId,
            generationId = group.generationId,
            protocolType = group.protocolType,
            protocolName = group.protocolName,
            leaderId = group.leaderOrNull, // 响应中包含Leader ID
            error = Errors.NONE) // 默认错误吗为NONE
          group.maybeInvokeJoinCallback(member, joinResult) // 调用回调以返回响应
          completeAndScheduleNextHeartbeatExpiration(group, member)
          member.isNew = false
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
// 返回响应时的回调
def sendResponseCallback(syncGroupResult: SyncGroupResult): Unit = {
  // 直接将SYNC_GROUP结果返回给客户端
  sendResponseMaybeThrottle(request, requestThrottleMs =>
    new SyncGroupResponse(
      new SyncGroupResponseData()
        .setErrorCode(syncGroupResult.error.code)
        .setProtocolType(syncGroupResult.protocolType.orNull)
        .setProtocolName(syncGroupResult.protocolName.orNull)
        .setAssignment(syncGroupResult.memberAssignment) // 分区分配信息
        .setThrottleTimeMs(requestThrottleMs)
    ))
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
private def doSyncGroup(group: GroupMetadata,
                        generationId: Int,
                        memberId: String,
                        protocolType: Option[String],
                        protocolName: Option[String],
                        groupInstanceId: Option[String],
                        groupAssignment: Map[String, Array[Byte]],
                        responseCallback: SyncCallback): Unit = {
  group.inLock {
      // ...
      group.currentState match {
        case Empty =>
          responseCallback(SyncGroupResult(Errors.UNKNOWN_MEMBER_ID))
        case PreparingRebalance =>
          responseCallback(SyncGroupResult(Errors.REBALANCE_IN_PROGRESS))
        case CompletingRebalance =>
          group.get(memberId).awaitingSyncCallback = responseCallback
          if (group.isLeader(memberId)) {
            // 对于主消费者而言，此时分配结果协调者已经知道
            val missing = group.allMembers -- groupAssignment.keySet
            val assignment = groupAssignment ++ missing.map(_ -> Array.empty[Byte]).toMap
            // ...
            groupManager.storeGroup(group, assignment, (error: Errors) => {
              group.inLock {
                if (group.is(CompletingRebalance) && generationId == group.generationId) {
                  if (error != Errors.NONE) {
                    // ... 
                  } else {
                    // 传播消费组的分配结果，并持久化到内部topic中
                    setAndPropagateAssignment(group, assignment)
                    // 将状态转成Stable
                    group.transitionTo(Stable)
                  }
                }
              }
            })
            // ...
          }
          // 而对于其他消费者，不会立刻返回响应，而是：
          // 1. 由上面的Leader消费者的请求触发，传播给其他消费者，此时消费组状态为Stable
          // 2. 由下面Stable状态的流程触发，直接返回分配信息

        case Stable =>
          // 若稳定，直接返回分配结果
          val memberMetadata = group.get(memberId)
          responseCallback(SyncGroupResult(group.protocolType, group.protocolName, memberMetadata.assignment, Errors.NONE))
          completeAndScheduleNextHeartbeatExpiration(group, group.get(memberId))
        case Dead =>
          throw new IllegalStateException(s"Reached unexpected condition for Dead group ${group.groupId}")
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private def setAndPropagateAssignment(group: GroupMetadata, assignment: Map[String, Array[Byte]]): Unit = {
  assert(group.is(CompletingRebalance))
  // 设置每个成员的分配结果
  group.allMemberMetadata.foreach(member => member.assignment = assignment(member.memberId))
  // 返回分配结果给每个成员
  propagateAssignment(group, Errors.NONE) 
}
private def propagateAssignment(group: GroupMetadata, error: Errors): Unit = {
  val (protocolType, protocolName) = if (error == Errors.NONE)
    (group.protocolType, group.protocolName)
  else
    (None, None)
  for (member <- group.allMemberMetadata) {
    // 对于每个成员，调用回调函数，将结果返回给每个成员
    if (group.maybeInvokeSyncCallback(member, SyncGroupResult(protocolType, protocolName, member.assignment, error))) {
      completeAndScheduleNextHeartbeatExpiration(group, member)
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def storeGroup(group: GroupMetadata,
               groupAssignment: Map[String, Array[Byte]],
               responseCallback: Errors => Unit): Unit = {
  getMagic(partitionFor(group.groupId)) match {
    case Some(magicValue) =>
      // ... 
      val groupMetadataRecords = // ... 需要保存到记录
      // 将记录保存到内部topic后，调用该回调将其存入缓存中
      def putCacheCallback(responseStatus: Map[TopicPartition, PartitionResponse]): Unit = {
        // ...
        val status = responseStatus(groupMetadataPartition)
        val responseError = if (status.error == Errors.NONE) {
          Errors.NONE
        } else {
          // ...
          }
        }
        // 调用回调responseCallback
        responseCallback(responseError)
      }
      // 将分配信息记录追加到内部topic中，追加完后，执行回调putCacheCallback
      appendForGroup(group, groupMetadataRecords, putCacheCallback)
    case None =>
      responseCallback(Errors.NOT_COORDINATOR)
      None
  }
}

private def appendForGroup(group: GroupMetadata,
                           records: Map[TopicPartition, MemoryRecords],
                           callback: Map[TopicPartition, PartitionResponse] => Unit): Unit = {
  // 将分配信息记录追加到内部topic中
  replicaManager.appendRecords(
    timeout = config.offsetCommitTimeoutMs.toLong,
    requiredAcks = config.offsetCommitRequiredAcks,
    internalTopicsAllowed = true,
    origin = AppendOrigin.Coordinator,
    entriesPerPartition = records,
    delayedProduceLock = Some(group.lock),
    responseCallback = callback)  // 执行完成后，运行回调，即上面的putCacheCallback
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private def maybePrepareRebalance(group: GroupMetadata, reason: String): Unit = {
  group.inLock { // 上锁
    if (group.canRebalance) // 这里状态必须是Stable, CompletingRebalance, Empty
      prepareRebalance(group, reason) // 开始准备“再平衡”
  }
}

private def prepareRebalance(group: GroupMetadata, reason: String): Unit = {
  if (group.is(CompletingRebalance)) // 但这里，若状态为CompletingRebalance，还是返回错误
                                     // 所以，成员状态必须时Stable或Empty，才能出发“再平衡”准备
    resetAndPropagateAssignmentError(group, Errors.REBALANCE_IN_PROGRESS)
  // 创建DelayJoin
  // 若目前状态为Empty，则创建InitialDelayedJoin；若为Stable，则创建DelayedJoin
  val delayedRebalance = if (group.is(Empty))
    new InitialDelayedJoin(this,
      joinPurgatory,
      group,
      groupConfig.groupInitialRebalanceDelayMs,
      groupConfig.groupInitialRebalanceDelayMs,
      max(group.rebalanceTimeoutMs - groupConfig.groupInitialRebalanceDelayMs, 0))
  else
    new DelayedJoin(this, group, group.rebalanceTimeoutMs)
  // 转换组状态为PreparingRebalance
  group.transitionTo(PreparingRebalance)
  val groupKey = GroupKey(group.groupId)
  // 创建完DelayedJoin操作后，立即尝试，看看是否能立即完成
  // 若不能立即完成，则会将操作加入到延迟缓存中（指定键为组ID）
  joinPurgatory.tryCompleteElseWatch(delayedRebalance, Seq(groupKey))
}

1
2
3
4
5
6
7
8
9
10
private[group] class DelayedJoin(coordinator: GroupCoordinator,
                                 group: GroupMetadata,
                                 rebalanceTimeout: Long) extends DelayedOperation(rebalanceTimeout, Some(group.lock)) {
  // 利用GroupCoordinator判断是否可以完成延迟入组，依据是所有消费者都发送了“加入组”请求
  override def tryComplete(): Boolean = coordinator.tryCompleteJoin(group, forceComplete _)
  // 任务过期的回调，定义在GroupCoordinator#onExpirationJoin中
  override def onExpiration() = coordinator.onExpireJoin()
  // 任务完成的回调，定义在GroupCoordinator#onCompleteJoin中
  override def onComplete() = coordinator.onCompleteJoin(group)
}

1
2
3
4
5
6
7
8
9
10
11
12
// In GroupCoordinator
def tryCompleteJoin(group: GroupMetadata, forceComplete: () => Boolean) = {
  group.inLock {
    if (group.hasAllMembersJoined) // 判断是否都发了“加入组”
      forceComplete() // 若都发了，则完成任务，最后会触发onComplete回调
    else false
  }
}

// In GroupMetadata
// 这里条件即：等待的成员个数等于组成员个数，且处于pending状态的为空
def hasAllMembersJoined = members.size == numMembersAwaitingJoin && pendingMembers.isEmpty

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def onCompleteJoin(group: GroupMetadata): Unit = {
  group.inLock { // 上锁
    // ... 
    } else {
      group.initNextGeneration() // 增加generation值，并将状态转化为CompletingRebalance
      // ...
      } else {
        // 对组内每个成员，发送JOIN_GROUP响应
        for (member <- group.allMemberMetadata) {
          val joinResult = JoinGroupResult(
            members = if (group.isLeader(member.memberId)) {
              group.currentMemberMetadata // 给Leader返回组成员信息
            } else {
              List.empty
            },
            memberId = member.memberId,
            generationId = group.generationId,
            protocolType = group.protocolType,
            protocolName = group.protocolName,
            leaderId = group.leaderOrNull,
            error = Errors.NONE)
					// 触发回调，即返回响应
          group.maybeInvokeJoinCallback(member, joinResult)
          completeAndScheduleNextHeartbeatExpiration(group, member)
          member.isNew = false
        }
      }
    }
  }
}

def maybeInvokeJoinCallback(member: MemberMetadata,
                            joinGroupResult: JoinGroupResult): Unit = {
  if (member.isAwaitingJoin) {
    member.awaitingJoinCallback(joinGroupResult) // 触发回调
    member.awaitingJoinCallback = null // 设置成员状态，成员不再awaitingJoin
    numMembersAwaitingJoin -= 1 // numMembersAwaitingJoin计数-1
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
private def doJoinGroup(...): Unit = {
  group.inLock { // 上锁
    // ...
    } else {
      val groupInstanceIdNotFound = groupInstanceId.isDefined && !group.hasStaticMember(groupInstanceId)
      // ...
      } else {
        val member = group.get(memberId)
        group.currentState match {
          // ...
          // 对应CompletingRebalance状态
          case CompletingRebalance =>
            if (member.matches(protocols)) {
              // 若元数据没改变，则不需要重分配，直接返回响应
              responseCallback(JoinGroupResult(...)
            } else {
              // 一般进入这里，执行4.1.～4.3.一样的操作
              updateMemberAndRebalance(group, member, protocols, responseCallback)
            }
					// ...
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
private def doSyncGroup(...): Unit = {
  group.inLock {
    // ...
    } else {
      group.currentState match {
        // ...
        // 此时之前的消费者SYNC_GROUP请求处理会来到这
        // 返回REBALANCE_IN_PROGRESS错误码
        // 消费者收到后会重新发送JOIN_GROUP请求
        case PreparingRebalance =>
          responseCallback(SyncGroupResult(Errors.REBALANCE_IN_PROGRESS))
        // ...
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
boolean joinGroupIfNeeded(final Timer timer) {
    while (rejoinNeededOrPending()) {
        // ...
        // 发起JOIN_GROUP，然后发起SYNC_GROUP
        // SYNC_GROUP会返回REBALANCE_IN_PROGRESS错误
        final RequestFuture<ByteBuffer> future = initiateJoinGroup();
        client.poll(future, timer); // 轮询以获取入组结果
        if (!future.isDone()) {
            return false; // 超时
        }
        if (future.succeeded()) {
            // ...
        } else {
            // 收到REBALANCE_IN_PROGRESS，Future会抛出RebanalceInProgress异常
            final RuntimeException exception = future.exception();
            // ...
            resetJoinGroupFuture(); // 重置（将joinFuture设为null，此时while条件返回true）
            if (exception instanceof UnknownMemberIdException ||
                exception instanceof RebalanceInProgressException || // 遇到RebanalceInProgress
                exception instanceof IllegalGenerationException ||
                exception instanceof MemberIdRequiredException)
                continue; // 进入下一轮循环，即重试，发起JOIN_GROUP请求
            else if (!future.isRetriable())
                throw exception;

            timer.sleep(rebalanceConfig.retryBackoffMs);
        }
    }
    return true;
}

protected synchronized boolean rejoinNeededOrPending() {
    return rejoinNeeded || joinFuture != null;
}

private synchronized void resetJoinGroupFuture() {
    this.joinFuture = null;
}