KafkaServer
对象中都会创建MetadataCache
组件, 负责缓存所有的metadata信息;val metadataCache: MetadataCache = new MetadataCache(config.brokerId)
PartitionStateInfo
private val cache: mutable.Map[String, mutable.Map[Int, PartitionStateInfo]] =
new mutable.HashMap[String, mutable.Map[Int, PartitionStateInfo]]()
PartitionStateInfo
: 包括LeaderIsrAndControllerEpoch
和Replica数组; 下面的readFrom
方法从接受到的buffer构造一个PartitionStateInfo
对象:def readFrom(buffer: ByteBuffer): PartitionStateInfo = {
val controllerEpoch = buffer.getInt
val leader = buffer.getInt
val leaderEpoch = buffer.getInt
val isrSize = buffer.getInt
val isr = for(i <- 0 until isrSize) yield buffer.getInt
val zkVersion = buffer.getInt
val replicationFactor = buffer.getInt
val replicas = for(i <- 0 until replicationFactor) yield buffer.getInt
PartitionStateInfo(LeaderIsrAndControllerEpoch(LeaderAndIsr(leader, leaderEpoch, isr.toList, zkVersion), controllerEpoch),
replicas.toSet)
}
MetadataCache
还保存着推送过来的有效的broker信息private var aliveBrokers: Map[Int, Broker] = Map()
KafkaApis
对象处理UpdateMetadataRequest
case RequestKeys.UpdateMetadataKey => handleUpdateMetadataRequest(request)
handleUpdateMetadataRequest
: val updateMetadataRequest = request.requestObj.asInstanceOf[UpdateMetadataRequest]
authorizeClusterAction(request)
replicaManager.maybeUpdateMetadataCache(updateMetadataRequest, metadataCache)
val updateMetadataResponse = new UpdateMetadataResponse(updateMetadataRequest.correlationId)
requestChannel.sendResponse(new Response(request, new RequestOrResponseSend(request.connectionId, updateMetadataResponse)))
可以看到是调用了ReplicaManager.maybeUpdateMetadataCache方法, 里面又会调用到MetadataCache.updateCache
方法
MetadataCache.updateCache
: aliveBrokers = updateMetadataRequest.aliveBrokers.map(b => (b.id, b)).toMap
updateMetadataRequest.partitionStateInfos.foreach { case(tp, info) =>
if (info.leaderIsrAndControllerEpoch.leaderAndIsr.leader == LeaderAndIsr.LeaderDuringDelete) {
removePartitionInfo(tp.topic, tp.partition)
} else {
addOrUpdatePartitionInfo(tp.topic, tp.partition, info)
}
}
干三件事
aliveBrokers
;removePartitionInfo(tp.topic, tp.partition)
;addOrUpdatePartitionInfo(tp.topic, tp.partition, info)
: 将信息meta信息保存到MetadataCache
的cache
对象中;UpdateMetaRequest
是谁在什么时候发送的;KafkaController
发送的;KafkaApis
对象处理MetadataRequest
case RequestKeys.MetadataKey => handleTopicMetadataRequest(request)
KafkaApis.handleTopicMetadataRequest
: val metadataRequest = request.requestObj.asInstanceOf[TopicMetadataRequest]
//if topics is empty -> fetch all topics metadata but filter out the topic response that are not authorized
val topics = if (metadataRequest.topics.isEmpty) {
val topicResponses = metadataCache.getTopicMetadata(metadataRequest.topics.toSet, request.securityProtocol)
topicResponses.map(_.topic).filter(topic => authorize(request.session, Describe, new Resource(Topic, topic))).toSet
} else {
metadataRequest.topics.toSet
}
//when topics is empty this will be a duplicate authorization check but given this should just be a cache lookup, it should not matter.
var (authorizedTopics, unauthorizedTopics) = topics.partition(topic => authorize(request.session, Describe, new Resource(Topic, topic)))
if (!authorizedTopics.isEmpty) {
val topicResponses = metadataCache.getTopicMetadata(authorizedTopics, request.securityProtocol)
if (config.autoCreateTopicsEnable && topicResponses.size != authorizedTopics.size) {
val nonExistentTopics: Set[String] = topics -- topicResponses.map(_.topic).toSet
authorizer.foreach {
az => if (!az.authorize(request.session, Create, Resource.ClusterResource)) {
authorizedTopics --= nonExistentTopics
unauthorizedTopics ++= nonExistentTopics
}
}
}
}
val unauthorizedTopicMetaData = unauthorizedTopics.map(topic => new TopicMetadata(topic, Seq.empty[PartitionMetadata], ErrorMapping.TopicAuthorizationCode))
val topicMetadata = if (authorizedTopics.isEmpty) Seq.empty[TopicMetadata] else getTopicMetadata(authorizedTopics, request.securityProtocol)
val brokers = metadataCache.getAliveBrokers
val response = new TopicMetadataResponse(brokers.map(_.getBrokerEndPoint(request.securityProtocol)), topicMetadata ++ unauthorizedTopicMetaData, metadataRequest.correlationId)
requestChannel.sendResponse(new RequestChannel.Response(request, new RequestOrResponseSend(request.connectionId, response)))
看着代码不少, 实际上比较简单:
authorizedTopics
和unauthorizedTopics
;getTopicMetadata
方法是关键所在, 它会先筛选出当前不存在的topic, 如果auto.create.topics.enable=true
, 则调用AdminUtils.createTopic
先创建此topic, 但此时其PartitionStateInfo为空, 不过也会作为Metadata Response的一部分返回给客户端.