[GitHub] [kafka] hachikuji 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)
hachikuji commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r683680618 ## File path: core/src/main/scala/kafka/log/Log.scala ## @@ -553,6 +553,17 @@ 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)) { +// we should never get here as the topic IDs should have been checked in becomeLeaderOrFollower Review comment: nit: I think it's ok to leave this out. The point of adding the check is to reduce the coupling with ReplicaManager ## File path: core/src/main/scala/kafka/log/Log.scala ## @@ -553,6 +553,17 @@ 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)) { +// we should never get here as the topic IDs should have been checked in becomeLeaderOrFollower +throw new InconsistentTopicIdException(s"Tried to assign topic ID $topicId to log for topic partition $topicPartition," + + s"but log already contained topic ID $current") + } + // Topic ID already assigned so we can return + return Review comment: I was thinking how we could avoid this return. How about something like this: ```scala _topicId match { case Some(currentId) => if (!currentId.equals(topicId)) { throw new InconsistentTopicIdException(s"Tried to assign topic ID $topicId to log for topic partition $topicPartition," + s"but log already contained topic ID $current") } case None => if (keepPartitionMetadataFile) { _topicId = Some(topicId) if (!partitionMetadataFile.exists()) { partitionMetadataFile.record(topicId) scheduler.schedule("flush-metadata-file", maybeFlushMetadataFile) } } } ``` -- 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] hachikuji 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)
hachikuji commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r683019148 ## 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: Can we shortcut return if the current topicId is already defined and matches the provided topicId? ## 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)) + // we should never get here as the topic IDs should have been checked in becomeLeaderOrFollower Review comment: nit: fix alignment (just use braces . I won't tell anyone) -- 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] hachikuji 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)
hachikuji commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r683019148 ## 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: Can we shortcut return if the current topicId is already defined and matches the provided topicId? ## 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)) + // we should never get here as the topic IDs should have been checked in becomeLeaderOrFollower Review comment: nit: fix alignment (just use braces . I won't tell anyone) -- 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] hachikuji 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)
hachikuji commented on a change in pull request #11171: URL: https://github.com/apache/kafka/pull/11171#discussion_r682206623 ## 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: By the time we get here, I think we have already validated that the topicid is consistent. Nevertheless, I wonder if it makes sense to let `assignTopicId` validate the passed topicId? Currently it will just override the value. ## 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 Review comment: Any better? ```scala def createLogIfNotExists(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints, topicId: Option[Uuid]): Unit = { def maybeCreate(logOpt: Option[Log]): Log = { logOpt match { case Some(log) => trace(s"${if (isFutureReplica) "Future Log" else "Log"} already exists.") topicId.foreach(log.assignTopicId) log case None => createLog(isNew, isFutureReplica, offsetCheckpoints, topicId) } } if (isFutureReplica) { this.futureLog = Some(maybeCreate(this.futureLog)) } else { this.log = Some(maybeCreate(this.log)) } } ``` -- 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