dajac commented on a change in pull request #8724:
URL: https://github.com/apache/kafka/pull/8724#discussion_r435085001
##########
File path: core/src/main/scala/kafka/controller/ControllerContext.scala
##########
@@ -391,6 +404,90 @@ class ControllerContext {
partitionsForTopic(topic).filter { partition =>
states.contains(partitionState(partition)) }.toSet
}
+ def putPartitionLeadershipInfo(partition: TopicPartition,
+ leaderIsrAndControllerEpoch:
LeaderIsrAndControllerEpoch): Unit = {
+ val previous = partitionLeadershipInfo.put(partition,
leaderIsrAndControllerEpoch)
+ val replicaAssignment = partitionFullReplicaAssignment(partition)
+ updatePreferredReplicaImbalanceMetric(partition, Some(replicaAssignment),
previous,
+ Some(replicaAssignment), Some(leaderIsrAndControllerEpoch))
+ }
+
+ def partitionLeadershipInfo(partition: TopicPartition):
Option[LeaderIsrAndControllerEpoch] = {
+ partitionLeadershipInfo.get(partition)
+ }
+
+ def partitionsLeadershipInfo(): Iterable[(TopicPartition,
LeaderIsrAndControllerEpoch)] = {
+ partitionLeadershipInfo
+ }
+
+ def partitionsWithLeaders(): Set[TopicPartition] = {
+ partitionLeadershipInfo.keySet
Review comment:
Right. I did not pay attention to this detail but that makes total sense
to use a better name here. Let's go with `partitionsWithOfflineLeaders`.
We can indeed exclude topics which are being queued for deletion. It is
already done in `partitionsWithOfflineLeaders` but I have added it in
`partitionsWithLeaders`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
