CalvinConfluent commented on code in PR #14612:
URL: https://github.com/apache/kafka/pull/14612#discussion_r1448299150
##########
core/src/main/scala/kafka/server/metadata/KRaftMetadataCache.scala:
##########
@@ -140,6 +141,71 @@ class KRaftMetadataCache(val brokerId: Int) extends
MetadataCache with Logging w
}
}
+ private def getPartitionMetadataForDescribeTopicResponse(
+ image: MetadataImage,
+ topicName: String,
+ listenerName: ListenerName
+ ): Option[List[DescribeTopicPartitionsResponsePartition]] = {
+ Option(image.topics().getTopic(topicName)) match {
+ case None => None
+ case Some(topic) => {
+ val partitions = Some(topic.partitions().entrySet().asScala.map {
entry =>
+ val partitionId = entry.getKey
+ val partition = entry.getValue
+ val filteredReplicas = maybeFilterAliveReplicas(image,
partition.replicas,
+ listenerName, false)
+ val filteredIsr = maybeFilterAliveReplicas(image, partition.isr,
listenerName,
+ false)
+ val offlineReplicas = getOfflineReplicas(image, partition,
listenerName)
+ val maybeLeader = getAliveEndpoint(image, partition.leader,
listenerName)
+ maybeLeader match {
+ case None =>
+ val error = if
(!image.cluster().brokers.containsKey(partition.leader)) {
+ debug(s"Error while fetching metadata for
$topicName-$partitionId: leader not available")
+ Errors.LEADER_NOT_AVAILABLE
+ } else {
+ debug(s"Error while fetching metadata for
$topicName-$partitionId: listener $listenerName " +
+ s"not found on leader ${partition.leader}")
+ Errors.LISTENER_NOT_FOUND
+ }
+ new DescribeTopicPartitionsResponsePartition()
+ .setErrorCode(error.code)
+ .setPartitionIndex(partitionId)
+ .setLeaderId(MetadataResponse.NO_LEADER_ID)
+ .setLeaderEpoch(partition.leaderEpoch)
+ .setReplicaNodes(filteredReplicas)
+ .setIsrNodes(filteredIsr)
+ .setOfflineReplicas(offlineReplicas)
+ case Some(leader) =>
+ val error = if (filteredReplicas.size <
partition.replicas.length) {
+ debug(s"Error while fetching metadata for
$topicName-$partitionId: replica information not available for " +
+ s"following brokers
${partition.replicas.filterNot(filteredReplicas.contains).mkString(",")}")
+ Errors.REPLICA_NOT_AVAILABLE
+ } else if (filteredIsr.size < partition.isr.length) {
+ debug(s"Error while fetching metadata for
$topicName-$partitionId: in sync replica information not available for " +
+ s"following brokers
${partition.isr.filterNot(filteredIsr.contains).mkString(",")}")
+ Errors.REPLICA_NOT_AVAILABLE
+ } else {
+ Errors.NONE
+ }
+
+ new DescribeTopicPartitionsResponsePartition()
+ .setErrorCode(error.code)
+ .setPartitionIndex(partitionId)
+ .setLeaderId(leader.id())
+ .setLeaderEpoch(partition.leaderEpoch)
+ .setReplicaNodes(filteredReplicas)
+ .setIsrNodes(filteredIsr)
+ .setOfflineReplicas(offlineReplicas)
+ .setEligibleLeaderReplicas(Replicas.toList(partition.elr))
+ .setLastKnownELR(Replicas.toList(partition.lastKnownElr))
+ }
+ }.toList)
+ partitions
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
