聊聊 Kafka：协调者 GroupCoordinator 源码剖析之实例化与启动 GroupCoordinator

class GroupMetadataManager(brokerId: Int,
                           interBrokerProtocolVersion: ApiVersion,
                           config: OffsetConfig,
                           // replicaManager 对象，管理 __consumer_offsets
                           val replicaManager: ReplicaManager,
                           zkClient: KafkaZkClient,
                           time: Time,
                           metrics: Metrics) extends Logging with KafkaMetricsGroup {
  private val compressionType: CompressionType = CompressionType.forId(config.offsetsTopicCompressionCodec.codec)
  // 记录每个 group 在服务端对应的 GroupMetadata 对象
  private val groupMetadataCache = new Pool[String, GroupMetadata]
  /* lock protecting access to loading and owned partition sets */
  private val partitionLock = new ReentrantLock()
  /* partitions of consumer groups that are being loaded, its lock should be always called BEFORE the group lock if needed */
  // 记录了正在加载的 offsets topic 分区的 ID
  private val loadingPartitions: mutable.Set[Int] = mutable.Set()
  /* partitions of consumer groups that are assigned, using the same loading partition lock */
  // 记录了已经加载的 offsets topic 分区的 ID
  private val ownedPartitions: mutable.Set[Int] = mutable.Set()
  /* shutting down flag */
  private val shuttingDown = new AtomicBoolean(false)
  /* number of partitions for the consumer metadata topic */
  // 记录 offsets topic 的分区数量，这个字段会调用 getGroupMetadataTopicPartitionCount() 进行初始化，默认 50。
  private val groupMetadataTopicPartitionCount = getGroupMetadataTopicPartitionCount
  /* single-thread scheduler to handle offset/group metadata cache loading and unloading */
  // 处理偏移、组元数据缓存加载和卸载
  private val scheduler = new KafkaScheduler(threads = 1, threadNamePrefix = "group-metadata-manager-")
  /* The groups with open transactional offsets commits per producer. We need this because when the commit or abort
   * marker comes in for a transaction, it is for a particular partition on the offsets topic and a particular producerId.
   * We use this structure to quickly find the groups which need to be updated by the commit/abort marker. */
  // 代表 transactional 有关的 producer 对应的 offset 记录
  private val openGroupsForProducer = mutable.HashMap[Long, mutable.Set[String]]()
  /* setup metrics*/
  private val partitionLoadSensor = metrics.sensor(GroupMetadataManager.LoadTimeSensor)
  ...
}

@nonthreadsafe
private[group] class GroupMetadata(val groupId: String, 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
  var protocolName: Option[String] = None
  var generationId = 0
  private var leaderId: Option[String] = None
  // 成员的集合
  private val members = new mutable.HashMap[String, MemberMetadata]
  // Static membership mapping [key: group.instance.id, value: member.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)
  // 每个 topic-partition 对应的 CommitRecordMetadataAndOffset (这个里面含有 offset 的 long 值、OffsetAndMetadata [offset 提交的时间戳、offset 超时的时间戳])
  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
  // When protocolType == `consumer`, a set of subscribed topics is maintained. The set is
  // computed when a new generation is created or when the group is restored from the log.
  private var subscribedTopics: Option[Set[String]] = None
  var newMemberAdded: Boolean = false
  ...
}