chb2ab commented on code in PR #14444: URL: https://github.com/apache/kafka/pull/14444#discussion_r1362351368
########## core/src/main/scala/kafka/server/KafkaApis.scala: ########## @@ -559,6 +560,26 @@ class KafkaApis(val requestChannel: RequestChannel, } } + case class LeaderNode(leaderId: Int, leaderEpoch: Int, node: Node) + + private def getCurrentLeader(tp: TopicPartition): LeaderNode = { + val partitionInfoOrError = replicaManager.getPartitionOrError(tp) + val (leaderId, leaderEpoch) = partitionInfoOrError match { + case Right(x) => + (x.leaderReplicaIdOpt.getOrElse(-1), x.getLeaderEpoch) + case Left(x) => + debug(s"Unable to retrieve local leaderId and Epoch with error $x, falling back to metadata cache") + metadataCache.getPartitionInfo(tp.topic, tp.partition) match { + case Some(pinfo) => (pinfo.leader(), pinfo.leaderEpoch()) + case None => (-1, -1) + } + } + val leaderNode: Node = metadataCache.getAliveBrokerNode(leaderId, config.interBrokerListenerName).getOrElse({ Review Comment: The metadata cache having more up to date information makes sense to me, but I don't follow the deletion case, would reading from the replica manager not return NOT_LEADER_OR_FOLLOWER there? -- 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