CalvinConfluent commented on code in PR #14612:
URL: https://github.com/apache/kafka/pull/14612#discussion_r1456867934
##########
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:
At least from the admin client perspective, when handling the
MetadataRequest(where this error handling copied from), partition level errors
are ignored. In our case, the DescribeTopicPartition request currently is only
used by the admin client, so I will remove the errors.
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java#L163
--
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]
