mumrah commented on code in PR #14612: URL: https://github.com/apache/kafka/pull/14612#discussion_r1453861304
########## 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)) { Review Comment: This won't cause an error for the whole request right? It will just populate the partition-level `ErrorCode` -- 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