CalvinConfluent commented on code in PR #14612:
URL: https://github.com/apache/kafka/pull/14612#discussion_r1446722719
##########
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:
Good question, actually I am not super sure about these errors. They may
have a history as they are included in the getPartitionMetadata method. @mumrah
Do you have an idea?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
