jolshan commented on code in PR #13111:
URL: https://github.com/apache/kafka/pull/13111#discussion_r1071407921
##########
core/src/main/scala/kafka/controller/KafkaController.scala:
##########
@@ -1716,27 +1726,25 @@ class KafkaController(val config: KafkaConfig,
}
if (!isActive) return
- val partitionReplicaAssignment =
zkClient.getFullReplicaAssignmentForTopics(immutable.Set(topic))
- val partitionsToBeAdded = partitionReplicaAssignment.filter { case
(topicPartition, _) =>
- controllerContext.partitionReplicaAssignment(topicPartition).isEmpty
- }
+ val partitions =
zkClient.getReplicaAssignmentAndTopicIdForTopics(immutable.Set(topic))
+ val partitionsToBeAdded = partitions.flatMap(_.assignment).filter(t =>
controllerContext.partitionReplicaAssignment(t._1).isEmpty)
if (topicDeletionManager.isTopicQueuedUpForDeletion(topic)) {
if (partitionsToBeAdded.nonEmpty) {
warn("Skipping adding partitions %s for topic %s since it is currently
being deleted"
.format(partitionsToBeAdded.map(_._1.partition).mkString(","),
topic))
- restorePartitionReplicaAssignment(topic, partitionReplicaAssignment)
+ restorePartitionReplicaAssignment(topic,
partitions.flatMap(_.assignment))
} else {
// This can happen if existing partition replica assignment are
restored to prevent increasing partition count during topic deletion
info("Ignoring partition change during topic deletion as no new
partitions are added")
}
} else if (partitionsToBeAdded.nonEmpty) {
info(s"New partitions to be added $partitionsToBeAdded")
- partitionsToBeAdded.forKeyValue { (topicPartition, assignedReplicas) =>
- controllerContext.updatePartitionFullReplicaAssignment(topicPartition,
assignedReplicas)
- }
- onNewPartitionCreation(partitionsToBeAdded.keySet)
+ processTopicIds(partitions)
+ partitionsToBeAdded.foreach(tuple =>
controllerContext.updatePartitionFullReplicaAssignment(tuple._1, tuple._2))
Review Comment:
I think switching this to "tuple" makes it harder to read and less clear
what the variables are.
--
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
