chb2ab commented on code in PR #14444:
URL: https://github.com/apache/kafka/pull/14444#discussion_r1362471338
##########
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:
ok, so my understanding is in the case of partition reassignments it would
be better to go directly to metadata cache, but when moving leadership within
the replica set it is better to go to replica manager first. I think we should
prioritize moving leadership within the replica set here since it seems more
common, what do you all think?
--
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
