KAFKA-12426: Missing logic to create partition.metadata files in RaftReplicaManager

core/src/main/scala/kafka/cluster/Partition.scala

@@ -425,36 +425,35 @@ class Partition(val topicPartition: TopicPartition,
   }
 
   /**
-   * Checks if the topic ID provided in the request is consistent with the topic ID in the log.
+   * Checks if the topic ID received is consistent with the topic ID in the log.
    * If a valid topic ID is provided, and the log exists but has no ID set, set the log ID to be the request ID.
    *
-   * @param requestTopicId the topic ID from the request
-   * @return true if the request topic id is consistent, false otherwise
+   * @param receivedTopicId the topic ID from the LeaderAndIsr request or from the metadata records
+   * @return true if the received topic id is consistent, false otherwise
    */
-  def checkOrSetTopicId(requestTopicId: Uuid): Boolean = {
-    // If the request had an invalid topic ID, then we assume that topic IDs are not supported.
-    // The topic ID was not inconsistent, so return true.
-    // If the log is empty, then we can not say that topic ID is inconsistent, so return true.
-    if (requestTopicId == null || requestTopicId == Uuid.ZERO_UUID)
-      true
-    else {
+  def checkOrSetTopicId(receivedTopicId: Uuid, usingRaft: Boolean): Boolean = {
+    // If the request had an invalid topic ID, then we assume that topic IDs are not supported so ID is consistent.
+    // This is only the case when from LeaderAndIsr Request. Raft code should never have invalid topic IDs.
+    if (receivedTopicId == null || receivedTopicId == Uuid.ZERO_UUID) {
+      !usingRaft // only acceptable when not using Raft
+    } else {
       log match {
-        case None => true
+        case None => true // log is empty, so we can not say topic ID is inconsistent
         case Some(log) => {
           // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file.
           // This is because if the broker previously wrote it to file, it would be recovered on restart after failure.
           // Topic ID is consistent since we are just setting it here.
           if (log.topicId == Uuid.ZERO_UUID) {
-            log.partitionMetadataFile.write(requestTopicId)
-            log.topicId = requestTopicId
+            log.partitionMetadataFile.write(receivedTopicId)
+            log.topicId = receivedTopicId
             true
-          } else if (log.topicId != requestTopicId) {
+          } else if (log.topicId != receivedTopicId) {
             stateChangeLogger.error(s"Topic Id in memory: ${log.topicId} does not" +
-              s" match the topic Id for partition $topicPartition provided in the request: " +
-              s"$requestTopicId.")
+              s" match the topic Id for partition $topicPartition provided in the " +
+              s"${if (usingRaft) "metadata records" else "request"}: $receivedTopicId.")
             false
           } else {
-            // topic ID in log exists and matches request topic ID
+            // topic ID in log exists and matches received topic ID
             true
           }
         }

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -1892,8 +1892,9 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
     }
   }
 
+  // Topic IDs are used with all self-managed quorum clusters and ZK cluster with IBP greater than or equal to 2.8
   def usesTopicId: Boolean =
-    interBrokerProtocolVersion >= KAFKA_2_8_IV0
+    usesSelfManagedQuorum || interBrokerProtocolVersion >= KAFKA_2_8_IV0
 
   validateValues()
 

core/src/main/scala/kafka/server/RaftReplicaManager.scala

@@ -26,8 +26,8 @@ import kafka.server.QuotaFactory.QuotaManagers
 import kafka.server.checkpoints.LazyOffsetCheckpoints
 import kafka.server.metadata.{ConfigRepository, MetadataImage, MetadataImageBuilder, MetadataPartition, RaftMetadataCache}
 import kafka.utils.Scheduler
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.errors.KafkaStorageException
+import org.apache.kafka.common.{TopicPartition, Uuid}
+import org.apache.kafka.common.errors.{InconsistentTopicIdException, KafkaStorageException}
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.utils.Time
 
@@ -251,6 +251,7 @@ class RaftReplicaManager(config: KafkaConfig,
               (Some(Partition(topicPartition, time, configRepository, this)), None)
           }
           partition.foreach { partition =>
+            checkOrSetTopicId(builder.topicNameToId(partition.topic), partition)
             val isNew = priorDeferredMetadata match {
               case Some(alreadyDeferred) => alreadyDeferred.isNew
               case _ => prevPartitions.topicPartition(topicPartition.topic(), topicPartition.partition()).isEmpty
@@ -286,6 +287,7 @@ class RaftReplicaManager(config: KafkaConfig,
               Some(partition)
           }
           partition.foreach { partition =>
+            checkOrSetTopicId(builder.topicNameToId(partition.topic), partition)
             if (currentState.leaderId == localBrokerId) {
               partitionsToBeLeader.put(partition, currentState)
             } else {
@@ -376,4 +378,15 @@ class RaftReplicaManager(config: KafkaConfig,
     metadataImage.partitions.topicPartition(partition.topic, partition.partitionId).getOrElse(
       throw new IllegalStateException(s"Partition has metadata changes but does not exist in the metadata cache: ${partition.topicPartition}"))
   }
+
+  private def checkOrSetTopicId(topicIdOpt: Option[Uuid], partition: Partition)= {
+     topicIdOpt match {
+      case Some(id) =>
+        if (!partition.checkOrSetTopicId(id, usingRaft = true))
+          throw new InconsistentTopicIdException(s"Topic partition ${partition.topicPartition} had an inconsistent topic ID." )
+      case None => throw new IllegalStateException(
+        s"Topic partition ${partition.topicPartition} is missing a topic ID"
+      )
+    }
+  }
 }

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -1369,7 +1369,7 @@ class ReplicaManager(val config: KafkaConfig,
               val requestLeaderEpoch = partitionState.leaderEpoch
               val requestTopicId = topicIds.get(topicPartition.topic)
 
-              if (!partition.checkOrSetTopicId(requestTopicId)) {
+              if (!partition.checkOrSetTopicId(requestTopicId, usingRaft = false)) {
                 responseMap.put(topicPartition, Errors.INCONSISTENT_TOPIC_ID)
               } else if (requestLeaderEpoch > currentLeaderEpoch) {
                 // If the leader epoch is valid record the epoch of the controller that made the leadership decision.

core/src/main/scala/kafka/server/metadata/MetadataImage.scala

@@ -47,6 +47,14 @@ case class MetadataImageBuilder(brokerId: Int,
     }
   }
 
+  def topicNameToId(topicName: String): Option[Uuid] = {
+    if (_partitionsBuilder != null) {
+      _partitionsBuilder.topicNameToId(topicName)
+    } else {
+      prevImage.topicNameToId(topicName)
+    }
+  }
+
   def controllerId(controllerId: Option[Int]): Unit = {
     _controllerId = controllerId
   }

core/src/test/scala/unit/kafka/server/RaftReplicaManagerTest.scala

@@ -23,8 +23,10 @@ import java.util.concurrent.atomic.AtomicBoolean
 
 import kafka.cluster.Partition
 import kafka.server.QuotaFactory.QuotaManagers
+import kafka.server.checkpoints.LazyOffsetCheckpoints
 import kafka.server.metadata.{CachedConfigRepository, MetadataBroker, MetadataBrokers, MetadataImage, MetadataImageBuilder, MetadataPartition, RaftMetadataCache}
 import kafka.utils.{MockScheduler, MockTime, TestUtils}
+import org.apache.kafka.common.errors.InconsistentTopicIdException
 import org.apache.kafka.common.{TopicPartition, Uuid}
 import org.apache.kafka.common.metadata.PartitionRecord
 import org.apache.kafka.common.metrics.Metrics
@@ -131,6 +133,11 @@ class RaftReplicaManagerTest {
     val partition0 =  Partition(topicPartition0, time, configRepository, rrm)
     val partition1 =  Partition(topicPartition1, time, configRepository, rrm)
 
+    rrm.createPartition(topicPartition0).createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+    rrm.createPartition(topicPartition1).createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+
     processTopicPartitionMetadata(rrm)
     // verify changes would have been deferred
     val partitionsNewMapCaptor: ArgumentCaptor[mutable.Map[Partition, Boolean]] =
@@ -161,6 +168,9 @@ class RaftReplicaManagerTest {
 
     // leadership change callbacks
     verifyLeadershipChangeCallbacks(List(partition0), List(partition1))
+
+    // partition.metadata file
+    verifyPartitionMetadataFile(rrm, List(topicPartition0, topicPartition1))
   }
 
   @Test
@@ -169,6 +179,12 @@ class RaftReplicaManagerTest {
     rrm.delegate = mockDelegate
     val partition0 = Partition(topicPartition0, time, configRepository, rrm)
     val partition1 = Partition(topicPartition1, time, configRepository, rrm)
+
+    rrm.createPartition(topicPartition0).createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+    rrm.createPartition(topicPartition1).createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+
     rrm.endMetadataChangeDeferral(onLeadershipChange)
 
     // define some return values to avoid NPE
@@ -188,6 +204,59 @@ class RaftReplicaManagerTest {
 
     // leadership change callbacks
     verifyLeadershipChangeCallbacks(List(partition0), List(partition1))
+
+    // partition.metadata file
+    verifyPartitionMetadataFile(rrm, List(topicPartition0, topicPartition1))
+  }
+
+  @Test
+  def testInconsistentTopicIdDefersChanges(): Unit = {
+    val rrm = createRaftReplicaManager()
+    rrm.delegate = mockDelegate
+    val partition0 = rrm.createPartition(topicPartition0)
+    val partition1 = rrm.createPartition(topicPartition1)
+
+    partition0.createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+    partition1.createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+
+    assertTrue(partition0.log.isDefined)
+    partition0.log.get.topicId = Uuid.randomUuid()
+
+    try {
+      processTopicPartitionMetadata(rrm)
+    } catch {
+      case e: Throwable => assertTrue(e.isInstanceOf[InconsistentTopicIdException])
+    }
+  }
+
+  @Test
+  def testInconsistentTopicIdWhenNotDeferring(): Unit = {
+    val rrm = createRaftReplicaManager()
+    rrm.delegate = mockDelegate
+    val partition0 = rrm.createPartition(topicPartition0)
+    val partition1 = rrm.createPartition(topicPartition1)
+
+    partition0.createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+    partition1.createLogIfNotExists(isNew = false, isFutureReplica = false,
+      new LazyOffsetCheckpoints(rrm.highWatermarkCheckpoints))
+
+    assertTrue(partition1.log.isDefined)
+    partition1.log.get.topicId = Uuid.randomUuid()
+
+    rrm.endMetadataChangeDeferral(onLeadershipChange)
+
+    // define some return values to avoid NPE
+    when(mockDelegate.makeLeaders(any(), any(), any(), ArgumentMatchers.eq(Some(offset1)))).thenReturn(Set(partition0))
+    when(mockDelegate.makeFollowers(any(), any(), any(), any(), ArgumentMatchers.eq(Some(offset1)))).thenReturn(Set(partition1))
+
+    try {
+      processTopicPartitionMetadata(rrm)
+    } catch {
+      case e: Throwable => assertTrue(e.isInstanceOf[InconsistentTopicIdException])
+    }
   }
 
   private def verifyMakeLeaders(expectedPrevPartitionsAlreadyExisting: Set[MetadataPartition],
@@ -221,6 +290,18 @@ class RaftReplicaManagerTest {
     assertEquals(expectedUpdatedFollowers, updatedFollowersCaptor.getValue.toList)
   }
 
+  private def verifyPartitionMetadataFile(rrm: RaftReplicaManager, topicPartitions: List[TopicPartition]) = {
+    topicPartitions.foreach ( topicPartition => {
+      val log = rrm.getLog(topicPartition).get
+      assertTrue(log.partitionMetadataFile.exists())
+      val partitionMetadata = log.partitionMetadataFile.read()
+
+      // Current version of PartitionMetadataFile is 0.
+      assertEquals(0, partitionMetadata.version)
+      assertEquals(topicId, partitionMetadata.topicId)
+    })
+  }
+
   private def processTopicPartitionMetadata(raftReplicaManager: RaftReplicaManager): Unit = {
     // create brokers
     imageBuilder.brokersBuilder().add(metadataBroker0)