core/src/main/scala/kafka/coordinator/MemberMetadata.scala
private var state: GroupState = Stable
coordinator
;private val members = new mutable.HashMap[String, MemberMetadata]
,
每个成员都有一个memberId, 对应着MemberMetadata
;var leaderId: String
: 对于group的balance, 简单来讲实际上是Coordinator
收集了所有的consumer的信息后, 将其发送给group中的一个consumer, 这个consumer负责按一定的balance策略,将partition分配到不同的consumer, 这个分配结果会Sync回Coordinator
, 然后再同步到各个consumer, 这个负责具体分配的consumer就是当前的Leader
; 这个Leader
的决定很简单, 谁第一个加入这个group的,谁就是leader;var protocol: String
: 当前group组所采用的balance策略, 选取的规则是被当前所有member都支持的策略中最多的那一个;var generationId
: 当前balance的一个标识id, 可以简单理解成是第几次作balance, 每次状态转换到AwaitingSync
时, 其值就增加1;core/src/main/scala/kafka/coordinator/GroupMetadataManager.scala
private val groupsCache = new Pool[String, GroupMetadata]
: 缓存了所有GroupMetadata
的信息;groupsCache
的管理接口:def getGroup(groupId: String): GroupMetadata
def addGroup(group: GroupMetadata): GroupMetadata
def removeGroup(group: GroupMetadata)
__consumer_offsets
, 写入其中的msg都包含有key
__consumer_offsets
这个topic里实际上保存两种类型消息:
2.1 一部分是offset信息(kafka.coordinator.OffsetsMessageFormatter
类型)的:
[groupId,topic,partition]::[OffsetMetadata[offset,metadata],CommitTime ExprirationTime]
, 它的key
是 [groupId,topic,partition]
2.2 另一部分是group信息(kafka.coordinator.GroupMetadataMessageFormatter
类型):
groupId::[groupId,Some(consumer),groupState,Map(memberId -> [memberId,clientId,clientHost,sessionTimeoutMs], ...->[]...)]
, 这部分实际上就是把当前Stable
状态的GroupMetadata
存到了__consumer_offsets
里, , 它的key
是 groupId
def store(delayedAppend: DelayedStore)
, 实现就是调用replicaManager.appendMessages
来写入消息到log文件__consumer_offsets
topic消息的加载 __consumer_offsets
作为一个topic, 也是有多个partiton的, 每个partiton也是有多个复本的, partition也会经历leader的选举,也会有故障转移操作;__consumer_offsets
在某台broker上的partition成为leader partition
时, 需要先从本地的log文件后加载offset,group相关信息到内存, 加载完成后才能对外提供读写和balance的操作;def loadGroupsForPartition(offsetsPartition: Int, onGroupLoaded: GroupMetadata => Unit)
private val offsetsCache = new Pool[GroupTopicPartition, OffsetAndMetadata]
;def getOffsets(group: String, topicPartitions: Seq[TopicAndPartition]): Map[TopicAndPartition, OffsetMetadataAndError]
offsetsCache
只保存最后一次提交的offset信息
private def putOffset(key: GroupTopicPartition, offsetAndMetadata: OffsetAndMetadata)
GroupMetadataManager
在启动时会同时启动一个名为delete-expired-consumer-offsets
定时任务来定时删除过期的offset信息;offsetsCache.remove(groupTopicAndPartition)
__consumer_offsets
的清除策略默认是compact
, 后面我们会单独开一章来讲日志的清除;core/src/main/scala/kafka/coordinator/GroupCoordinator.scala
case RequestKeys.OffsetCommitKey => handleOffsetCommitRequest(request)
case RequestKeys.OffsetFetchKey => handleOffsetFetchRequest(request)
case RequestKeys.GroupCoordinatorKey => handleGroupCoordinatorRequest(request)
case RequestKeys.JoinGroupKey => handleJoinGroupRequest(request)
case RequestKeys.HeartbeatKey => handleHeartbeatRequest(request)
case RequestKeys.LeaveGroupKey => handleLeaveGroupRequest(request)
case RequestKeys.SyncGroupKey => handleSyncGroupRequest(request)
case RequestKeys.DescribeGroupsKey => handleDescribeGroupRequest(request)
case RequestKeys.ListGroupsKey => handleListGroupsRequest(request)
c1.jpg
c2.jpg
JoinRequest
来触发新的balance;