[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)

[GitBox <gi...@apache.org>]

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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org