[GitHub] [kafka] jolshan commented on a change in pull request #11171: KAFKA-13132: Upgrading to topic IDs in LISR requests has gaps introduced in 3.0 (part 2)
jolshan commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r682739288 ## File path: core/src/main/scala/kafka/cluster/Partition.scala ## @@ -313,14 +313,20 @@ class Partition(val topicPartition: TopicPartition, } def createLogIfNotExists(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Unit = { -isFutureReplica match { - case true if futureLog.isEmpty => -val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) +val logOpt = if (isFutureReplica) futureLog else log +if (logOpt.isEmpty) { + val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) + if (isFutureReplica) this.futureLog = Option(log) - case false if log.isEmpty => -val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) + else this.log = Option(log) - case _ => trace(s"${if (isFutureReplica) "Future Log" else "Log"} already exists.") +} else { + trace(s"${if (isFutureReplica) "Future Log" else "Log"} already exists.") + logOpt.foreach { log => +if (log.topicId.isEmpty) { Review comment: I'm a little unsure what you mean by topicId is consistent here. (There would have been no log/topic ID in log to compare to). We can add an extra check in assign topic ID to ensure there is no way to assign inconsistently. -- 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
[GitHub] [kafka] jolshan commented on a change in pull request #11171: KAFKA-13132: Upgrading to topic IDs in LISR requests has gaps introduced in 3.0 (part 2)
jolshan commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r683019982 ## File path: core/src/main/scala/kafka/log/Log.scala ## @@ -553,6 +553,14 @@ class Log(@volatile var logStartOffset: Long, /** Only used for ZK clusters when we update and start using topic IDs on existing topics */ def assignTopicId(topicId: Uuid): Unit = { +// defensively check that any newly assign topic ID matches any that is already set +_topicId.foreach { current => + if (!current.equals(topicId)) Review comment: We can. I was just looking at this case again. -- 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
[GitHub] [kafka] jolshan commented on a change in pull request #11171: KAFKA-13132: Upgrading to topic IDs in LISR requests has gaps introduced in 3.0 (part 2)
jolshan commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r682739288 ## File path: core/src/main/scala/kafka/cluster/Partition.scala ## @@ -313,14 +313,20 @@ class Partition(val topicPartition: TopicPartition, } def createLogIfNotExists(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Unit = { -isFutureReplica match { - case true if futureLog.isEmpty => -val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) +val logOpt = if (isFutureReplica) futureLog else log +if (logOpt.isEmpty) { + val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) + if (isFutureReplica) this.futureLog = Option(log) - case false if log.isEmpty => -val log = createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) + else this.log = Option(log) - case _ => trace(s"${if (isFutureReplica) "Future Log" else "Log"} already exists.") +} else { + trace(s"${if (isFutureReplica) "Future Log" else "Log"} already exists.") + logOpt.foreach { log => +if (log.topicId.isEmpty) { Review comment: I'm a little unsure what you mean by topicId is consistent here. (There would have been no log/topic ID in log to compare to). We can add an extra check in assign topic ID to ensure there is no way to assign inconsistently. ## File path: core/src/main/scala/kafka/log/Log.scala ## @@ -553,6 +553,14 @@ class Log(@volatile var logStartOffset: Long, /** Only used for ZK clusters when we update and start using topic IDs on existing topics */ def assignTopicId(topicId: Uuid): Unit = { +// defensively check that any newly assign topic ID matches any that is already set +_topicId.foreach { current => + if (!current.equals(topicId)) Review comment: We can. I was just looking at this case again. -- 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