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

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

@@ -344,10 +344,6 @@ class Partition(val topicPartition: TopicPartition,
       val log = logManager.getOrCreateLog(topicPartition, isNew, isFutureReplica, topicId)
       maybeLog = Some(log)
       updateHighWatermark(log)
-      // When running a ZK controller, we may get a log that does not have a topic ID. Assign it here.
-      if (log.topicId == None) {
-        topicId.foreach(topicId => log.writeTopicIdToExistingLog(topicId))
-      }
       log
     } finally {
       logManager.finishedInitializingLog(topicPartition, maybeLog)

core/src/main/scala/kafka/log/Log.scala

@@ -267,7 +267,7 @@ class Log(@volatile private var _dir: File,
           logDirFailureChannel: LogDirFailureChannel,
           private val hadCleanShutdown: Boolean = true,
           @volatile var topicId: Option[Uuid],
-          val keepPartitionMetadataFile: Boolean = true) extends Logging with KafkaMetricsGroup {
+          val keepPartitionMetadataFile: Boolean) extends Logging with KafkaMetricsGroup {
 
   import kafka.log.Log._
 
@@ -365,8 +365,8 @@ class Log(@volatile private var _dir: File,
               s"but log already contained topic ID $fileTopicId")
           topicId = Some(fileTopicId)
         }
-    } else if (topicId.isDefined && keepPartitionMetadataFile) {
-      partitionMetadataFile.write(topicId.get)
+    } else if (keepPartitionMetadataFile) {
+      topicId.foreach(partitionMetadataFile.write)
     }
   }
 
@@ -590,7 +590,7 @@ class Log(@volatile private var _dir: File,
   }
 
   /** Only used for ZK clusters when we update and start using topic IDs on existing topics */
-  def writeTopicIdToExistingLog(topicId: Uuid): Unit = {
+  def assignTopicId(topicId: Uuid): Unit = {
     partitionMetadataFile.write(topicId)
     this.topicId = Some(topicId)
   }
@@ -2620,7 +2620,7 @@ object Log {
             logDirFailureChannel: LogDirFailureChannel,
             lastShutdownClean: Boolean = true,
             topicId: Option[Uuid],
-            keepPartitionMetadataFile: Boolean = true): Log = {
+            keepPartitionMetadataFile: Boolean): Log = {
     val topicPartition = Log.parseTopicPartitionName(dir)
     val producerStateManager = new ProducerStateManager(topicPartition, dir, maxProducerIdExpirationMs)
     new Log(dir, config, logStartOffset, recoveryPoint, scheduler, brokerTopicStats, time, maxProducerIdExpirationMs,

core/src/main/scala/kafka/log/LogManager.scala

@@ -776,12 +776,12 @@ class LogManager(logDirs: Seq[File],
    * @param topicPartition The partition whose log needs to be returned or created
    * @param isNew Whether the replica should have existed on the broker or not
    * @param isFuture True if the future log of the specified partition should be returned or created
-   * @param topicId The topic ID of the topic used in the case of log creation.
+   * @param topicId The topic ID of the partition's topic
    * @throws KafkaStorageException if isNew=false, log is not found in the cache and there is offline log directory on the broker
    */
   def getOrCreateLog(topicPartition: TopicPartition, isNew: Boolean = false, isFuture: Boolean = false, topicId: Option[Uuid]): Log = {
     logCreationOrDeletionLock synchronized {
-      getLog(topicPartition, isFuture).getOrElse {
+      val log = getLog(topicPartition, isFuture).getOrElse {
         // create the log if it has not already been created in another thread
         if (!isNew && offlineLogDirs.nonEmpty)
           throw new KafkaStorageException(s"Can not create log for $topicPartition because log directories ${offlineLogDirs.mkString(",")} are offline")
@@ -842,6 +842,20 @@ class LogManager(logDirs: Seq[File],
 
         log
       }
+      // When running a ZK controller, we may get a log that does not have a topic ID. Assign it here.
+      if (log.topicId.isEmpty) {
+        topicId.foreach(log.assignTopicId)
+      }
+
+      // Ensure topic IDs are consistent
+      topicId.foreach { topicId =>
+        log.topicId.foreach { logTopicId =>
+          if (topicId != logTopicId)
+            throw new IllegalStateException(s"Tried to assign topic ID $topicId to log for topic partition $topicPartition," +
+              s"but log already contained topic ID $logTopicId")
+        }
+      }
+      log
     }
   }
 

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

@@ -389,15 +389,15 @@ class RaftReplicaManager(config: KafkaConfig,
   private def checkTopicId(receivedTopicIdOpt: Option[Uuid], logTopicIdOpt: Option[Uuid], topicPartition: TopicPartition): Unit = {
     receivedTopicIdOpt match {
       case Some(receivedTopicId) =>
-        logTopicIdOpt.foreach(logTopicId => {
+        logTopicIdOpt.foreach { logTopicId =>
           if (receivedTopicId != logTopicId) {
             // not sure if we need both the logger and the error thrown
             stateChangeLogger.error(s"Topic ID in memory: $logTopicId does not" +
               s" match the topic ID for partition $topicPartition received: " +
               s"$receivedTopicId.")
             throw new InconsistentTopicIdException(s"Topic partition $topicPartition had an inconsistent topic ID.")
           }
-        })
+        }
       case None => throw new IllegalStateException(s"Topic partition $topicPartition is missing a topic ID")
     }
   }

core/src/test/scala/other/kafka/StressTestLog.scala

@@ -52,7 +52,8 @@ object StressTestLog {
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
       brokerTopicStats = new BrokerTopicStats,
       logDirFailureChannel = new LogDirFailureChannel(10),
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
     val writer = new WriterThread(log)
     writer.start()
     val reader = new ReaderThread(log)

core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala

@@ -211,7 +211,7 @@ object TestLinearWriteSpeed {
   class LogWritable(val dir: File, config: LogConfig, scheduler: Scheduler, val messages: MemoryRecords) extends Writable {
     Utils.delete(dir)
     val log = Log(dir, config, 0L, 0L, scheduler, new BrokerTopicStats, Time.SYSTEM, 60 * 60 * 1000,
-      LogManager.ProducerIdExpirationCheckIntervalMs, new LogDirFailureChannel(10), topicId = None)
+      LogManager.ProducerIdExpirationCheckIntervalMs, new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true)
     def write(): Int = {
       log.appendAsLeader(messages, leaderEpoch = 0)
       messages.sizeInBytes

core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala

@@ -354,7 +354,8 @@ class PartitionLockTest extends Logging {
     log.topicPartition,
     log.producerStateManager,
     new LogDirFailureChannel(1),
-    topicId = None) {
+    topicId = None,
+    keepPartitionMetadataFile = true) {
 
     override def appendAsLeader(records: MemoryRecords, leaderEpoch: Int, origin: AppendOrigin, interBrokerProtocolVersion: ApiVersion): LogAppendInfo = {
       val appendInfo = super.appendAsLeader(records, leaderEpoch, origin, interBrokerProtocolVersion)

core/src/test/scala/unit/kafka/cluster/PartitionTest.scala

@@ -1672,9 +1672,12 @@ class PartitionTest extends AbstractPartitionTest {
     assertEquals(topicId, partition2.topicId.get)
     assertFalse(partition2.log.isDefined)
 
-    // Calling makeLeader with a new topic ID should not overwrite the old topic ID. We should get the same log.
-    // This scenario should not occur, since the topic ID check will fail, but it is good to check we grab the old log.
-    partition2.makeLeader(leaderState, offsetCheckpoints, Some(Uuid.randomUuid()))
+    // Calling makeLeader with a new topic ID should not overwrite the old topic ID. We should get an IllegalStateException.
+    // This scenario should not occur, since the topic ID check will fail.
+    assertThrows(classOf[IllegalStateException], () => partition2.makeLeader(leaderState, offsetCheckpoints, Some(Uuid.randomUuid())))
+
+    // Calling makeLeader with no topic ID should not overwrite the old topic ID. We should get the original log.
+    partition2.makeLeader(leaderState, offsetCheckpoints, None)
     checkTopicId(topicId, partition2)
   }
 
@@ -1713,9 +1716,12 @@ class PartitionTest extends AbstractPartitionTest {
     assertEquals(topicId, partition2.topicId.get)
     assertFalse(partition2.log.isDefined)
 
-    // Calling makeFollower with a new topic ID should not overwrite the old topic ID. We should get the same log.
-    // This scenario should not occur, since the topic ID check will fail, but it is good to check we grab the old log.
-    partition2.makeFollower(leaderState, offsetCheckpoints, Some(Uuid.randomUuid()))
+    // Calling makeFollower with a new topic ID should not overwrite the old topic ID. We should get an IllegalStateException.
+    // This scenario should not occur, since the topic ID check will fail.
+    assertThrows(classOf[IllegalStateException], () => partition2.makeFollower(leaderState, offsetCheckpoints, Some(Uuid.randomUuid())))
+
+    // Calling makeFollower with no topic ID should not overwrite the old topic ID. We should get the original log.
+    partition2.makeFollower(leaderState, offsetCheckpoints, None)
     checkTopicId(topicId, partition2)
   }
 
@@ -1941,7 +1947,8 @@ class PartitionTest extends AbstractPartitionTest {
     log.topicPartition,
     log.producerStateManager,
     new LogDirFailureChannel(1),
-    topicId = None) {
+    topicId = None,
+    keepPartitionMetadataFile = true) {
 
     override def appendAsFollower(records: MemoryRecords): LogAppendInfo = {
       appendSemaphore.acquire()

core/src/test/scala/unit/kafka/cluster/ReplicaTest.scala

@@ -51,7 +51,8 @@ class ReplicaTest {
       maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
       logDirFailureChannel = new LogDirFailureChannel(10),
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
   }
 
   @AfterEach

core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala

@@ -111,7 +111,8 @@ abstract class AbstractLogCleanerIntegrationTest {
         maxProducerIdExpirationMs = 60 * 60 * 1000,
         producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
         logDirFailureChannel = new LogDirFailureChannel(10),
-        topicId = None)
+        topicId = None,
+        keepPartitionMetadataFile = true)
       logMap.put(partition, log)
       this.logs += log
     }

core/src/test/scala/unit/kafka/log/BrokerCompressionTest.scala

@@ -55,7 +55,7 @@ class BrokerCompressionTest {
     val log = Log(logDir, LogConfig(logProps), logStartOffset = 0L, recoveryPoint = 0L, scheduler = time.scheduler,
       time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
-      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None)
+      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true)
 
     /* append two messages */
     log.appendAsLeader(MemoryRecords.withRecords(CompressionType.forId(messageCompressionCode.codec), 0,

core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala

@@ -98,7 +98,8 @@ class LogCleanerManagerTest extends Logging {
     // the exception should be catched and the partition that caused it marked as uncleanable
     class LogMock(dir: File, config: LogConfig) extends Log(dir, config, 0L, 0L,
       time.scheduler, new BrokerTopicStats, time, 60 * 60 * 1000, LogManager.ProducerIdExpirationCheckIntervalMs,
-      topicPartition, new ProducerStateManager(tp, tpDir, 60 * 60 * 1000), new LogDirFailureChannel(10), topicId = None) {
+      topicPartition, new ProducerStateManager(tp, tpDir, 60 * 60 * 1000),
+      new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true) {
 
       // Throw an error in getFirstBatchTimestampForSegments since it is called in grabFilthiestLog()
       override def getFirstBatchTimestampForSegments(segments: Iterable[LogSegment]): Iterable[Long] =
@@ -756,7 +757,8 @@ class LogCleanerManagerTest extends Logging {
       maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
       logDirFailureChannel = new LogDirFailureChannel(10),
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
   }
 
   private def createLowRetentionLogConfig(segmentSize: Int, cleanupPolicy: String): LogConfig = {
@@ -800,7 +802,7 @@ class LogCleanerManagerTest extends Logging {
     Log(dir = dir, config = config, logStartOffset = 0L, recoveryPoint = 0L, scheduler = time.scheduler,
       time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
-      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None)
+      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true)
 
   private def records(key: Int, value: Int, timestamp: Long) =
     MemoryRecords.withRecords(CompressionType.NONE, new SimpleRecord(timestamp, key.toString.getBytes, value.toString.getBytes))

core/src/test/scala/unit/kafka/log/LogCleanerTest.scala

@@ -113,7 +113,8 @@ class LogCleanerTest {
                       topicPartition = topicPartition,
                       producerStateManager = producerStateManager,
                       logDirFailureChannel = new LogDirFailureChannel(10),
-                      topicId = None) {
+                      topicId = None,
+                      keepPartitionMetadataFile = true) {
       override def replaceSegments(newSegments: Seq[LogSegment], oldSegments: Seq[LogSegment], isRecoveredSwapFile: Boolean = false): Unit = {
         deleteStartLatch.countDown()
         if (!deleteCompleteLatch.await(5000, TimeUnit.MILLISECONDS)) {
@@ -1675,7 +1676,7 @@ class LogCleanerTest {
     Log(dir = dir, config = config, logStartOffset = 0L, recoveryPoint = recoveryPoint, scheduler = time.scheduler,
       time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
-      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None)
+      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true)
 
   private def makeCleaner(capacity: Int, checkDone: TopicPartition => Unit = _ => (), maxMessageSize: Int = 64*1024) =
     new Cleaner(id = 0,

core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala

@@ -151,7 +151,8 @@ class LogConcurrencyTest {
       maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
       logDirFailureChannel = new LogDirFailureChannel(10),
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
   }
 
   private def validateConsumedData(log: Log, consumedBatches: Iterable[FetchedBatch]): Unit = {

core/src/test/scala/unit/kafka/log/LogTest.scala

@@ -110,7 +110,7 @@ class LogTest {
 
           val producerStateManager = new ProducerStateManager(topicPartition, logDir, maxPidExpirationMs)
           val log = new Log(logDir, config, logStartOffset, logRecoveryPoint, time.scheduler, brokerTopicStats, time, maxPidExpirationMs,
-            LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, hadCleanShutdown, None) {
+            LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition, producerStateManager, logDirFailureChannel, hadCleanShutdown, None, true) {
             override def recoverLog(): Long = {
               if (simulateError)
                 throw new RuntimeException
@@ -1025,7 +1025,7 @@ class LogTest {
       // Intercept all segment read calls
       new Log(logDir, logConfig, logStartOffset = 0, recoveryPoint = recoveryPoint, mockTime.scheduler,
         brokerTopicStats, mockTime, maxProducerIdExpirationMs, LogManager.ProducerIdExpirationCheckIntervalMs,
-        topicPartition, producerStateManager, new LogDirFailureChannel(10), hadCleanShutdown = false, topicId = None) {
+        topicPartition, producerStateManager, new LogDirFailureChannel(10), hadCleanShutdown = false, topicId = None, keepPartitionMetadataFile = true) {
 
         override def addSegment(segment: LogSegment): LogSegment = {
           val wrapper = new LogSegment(segment.log, segment.lazyOffsetIndex, segment.lazyTimeIndex, segment.txnIndex, segment.baseOffset,
@@ -1129,7 +1129,8 @@ class LogTest {
       producerStateManager = stateManager,
       logDirFailureChannel = new LogDirFailureChannel(1),
       hadCleanShutdown = false,
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
 
@@ -1208,7 +1209,8 @@ class LogTest {
       topicPartition = Log.parseTopicPartitionName(logDir),
       producerStateManager = stateManager,
       logDirFailureChannel = null,
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
   }
@@ -1247,7 +1249,8 @@ class LogTest {
       topicPartition = Log.parseTopicPartitionName(logDir),
       producerStateManager = stateManager,
       logDirFailureChannel = null,
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
   }
@@ -1288,7 +1291,8 @@ class LogTest {
       topicPartition = Log.parseTopicPartitionName(logDir),
       producerStateManager = stateManager,
       logDirFailureChannel = null,
-      topicId = None)
+      topicId = None,
+      keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
   }
@@ -4959,7 +4963,8 @@ object LogTest {
       producerIdExpirationCheckIntervalMs = producerIdExpirationCheckIntervalMs,
       logDirFailureChannel = new LogDirFailureChannel(10),
       lastShutdownClean = lastShutdownClean,
-      topicId = topicId)
+      topicId = topicId,
+      keepPartitionMetadataFile = true)
   }
 
   /**

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

@@ -1495,7 +1495,8 @@ class ReplicaManagerTest {
       producerStateManager = new ProducerStateManager(new TopicPartition(topic, topicPartition),
         new File(new File(config.logDirs.head), s"$topic-$topicPartition"), 30000),
       logDirFailureChannel = mockLogDirFailureChannel,
-      topicId = topicId) {
+      topicId = topicId,
+      keepPartitionMetadataFile = true) {
 
       override def endOffsetForEpoch(leaderEpoch: Int): Option[OffsetAndEpoch] = {
         assertEquals(leaderEpoch, leaderEpochFromLeader)
@@ -2315,7 +2316,7 @@ class ReplicaManagerTest {
   }
 
   @Test
-  def testInvalidIdReturnsError(): Unit = {
+  def testInconsistentIdReturnsError(): Unit = {
     val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time))
     try {
       val brokerList = Seq[Integer](0, 1).asJava
@@ -2347,7 +2348,7 @@ class ReplicaManagerTest {
       val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, topicIds), (_, _) => ())
       assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition))
 
-      // Send request with invalid ID.
+      // Send request with inconsistent ID.
       val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, invalidTopicIds), (_, _) => ())
       assertEquals(Errors.INCONSISTENT_TOPIC_ID, response3.partitionErrors(invalidTopicNames).get(topicPartition))
 

core/src/test/scala/unit/kafka/tools/DumpLogSegmentsTest.scala

@@ -61,7 +61,7 @@ class DumpLogSegmentsTest {
     log = Log(logDir, LogConfig(props), logStartOffset = 0L, recoveryPoint = 0L, scheduler = time.scheduler,
       time = time, brokerTopicStats = new BrokerTopicStats, maxProducerIdExpirationMs = 60 * 60 * 1000,
       producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs,
-      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None)
+      logDirFailureChannel = new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true)
   }
 
   def addSimpleRecords(): Unit = {

core/src/test/scala/unit/kafka/utils/SchedulerTest.scala

@@ -124,7 +124,7 @@ class SchedulerTest {
     val producerStateManager = new ProducerStateManager(topicPartition, logDir, maxProducerIdExpirationMs)
     val log = new Log(logDir, logConfig, logStartOffset = 0, recoveryPoint = recoveryPoint, scheduler,
       brokerTopicStats, mockTime, maxProducerIdExpirationMs, LogManager.ProducerIdExpirationCheckIntervalMs,
-      topicPartition, producerStateManager, new LogDirFailureChannel(10), topicId = None)
+      topicPartition, producerStateManager, new LogDirFailureChannel(10), topicId = None, keepPartitionMetadataFile = true)
     assertTrue(scheduler.taskRunning(log.producerExpireCheck))
     log.close()
     assertFalse(scheduler.taskRunning(log.producerExpireCheck))