jolshan commented on code in PR #13111:
URL: https://github.com/apache/kafka/pull/13111#discussion_r1071421020
##########
core/src/main/scala/kafka/controller/KafkaController.scala:
##########
@@ -1664,11 +1664,21 @@ class KafkaController(val config: KafkaConfig,
}
private def processTopicIds(topicIdAssignments:
Set[TopicIdReplicaAssignment]): Unit = {
- // Create topic IDs for topics missing them if we are using topic IDs
+ // Create topic IDs or update with locally stored topicIDs for topics
missing them if we are using topic IDs
// Otherwise, maintain what we have in the topicZNode
val updatedTopicIdAssignments = if (config.usesTopicId) {
- val (withTopicIds, withoutTopicIds) =
topicIdAssignments.partition(_.topicId.isDefined)
- withTopicIds ++ zkClient.setTopicIds(withoutTopicIds,
controllerContext.epochZkVersion)
+ val (withTopicIds, withoutTopicIds, withLocalTopicIds) =
topicIdAssignments.foldLeft((Set.empty[TopicIdReplicaAssignment],
Set.empty[TopicIdReplicaAssignment], Set.empty[TopicIdReplicaAssignment])) {
Review Comment:
Since we go through all the partitions again to check the topic ID and add
to the replica assignment I wonder if we could have kept the partition and then
a separate part for the ones with local IDs. 🤔
--
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
