MINOR: Don't ignore deletion of partition metadata file and log topic id clean-ups

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

@@ -272,7 +272,7 @@ class Log(@volatile private var _dir: File,
           @volatile var leaderEpochCache: Option[LeaderEpochFileCache],
           val producerStateManager: ProducerStateManager,
           logDirFailureChannel: LogDirFailureChannel,
-          @volatile var topicId: Option[Uuid],
+          @volatile private var _topicId: Option[Uuid],
           val keepPartitionMetadataFile: Boolean) extends Logging with KafkaMetricsGroup {
 
   import kafka.log.Log._
@@ -326,18 +326,24 @@ class Log(@volatile private var _dir: File,
     if (partitionMetadataFile.exists()) {
         if (keepPartitionMetadataFile) {
           val fileTopicId = partitionMetadataFile.read().topicId
-          if (topicId.isDefined && !topicId.contains(fileTopicId))
+          if (_topicId.isDefined && !_topicId.contains(fileTopicId))
             throw new InconsistentTopicIdException(s"Tried to assign topic ID $topicId to log for topic partition $topicPartition," +
               s"but log already contained topic ID $fileTopicId")
-          topicId = Some(fileTopicId)
+          _topicId = Some(fileTopicId)
         } else {
-          partitionMetadataFile.delete()
+          try partitionMetadataFile.delete()
+          catch {
+            case e: IOException =>
+              error(s"Error while trying to delete partition metadata file ${partitionMetadataFile}", e)
+          }
         }
     } else if (keepPartitionMetadataFile) {
-      topicId.foreach(partitionMetadataFile.write)
+      _topicId.foreach(partitionMetadataFile.write)
     }
   }
 
+  def topicId: Option[Uuid] = _topicId
+
   def dir: File = _dir
 
   def parentDir: String = _parentDir
@@ -553,7 +559,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 assignTopicId(topicId: Uuid): Unit = {
     partitionMetadataFile.write(topicId)
-    this.topicId = Some(topicId)
+    _topicId = Some(topicId)
   }
 
   private def initializeLeaderEpochCache(): Unit = lock synchronized {

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

@@ -138,7 +138,9 @@ class PartitionMetadataFile(val file: File,
     file.exists()
   }
 
-  def delete(): Boolean = {
-    file.delete()
+  def delete(): Unit = {
+    Files.delete(file.toPath)
   }
+
+  override def toString: String = s"PartitionMetadataFile(path=$path)"
 }

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

@@ -385,7 +385,7 @@ class PartitionLockTest extends Logging {
     leaderEpochCache,
     producerStateManager,
     logDirFailureChannel,
-    topicId = None,
+    _topicId = None,
     keepPartitionMetadataFile = true) {
 
     override def appendAsLeader(records: MemoryRecords, leaderEpoch: Int, origin: AppendOrigin, interBrokerProtocolVersion: ApiVersion): LogAppendInfo = {

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

@@ -72,7 +72,7 @@ class PartitionTest extends AbstractPartitionTest {
     assertEquals(17L, log.logEndOffset)
 
     val leaderEpoch = 10
-    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true, log = log)
+    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true)
 
     def epochEndOffset(epoch: Int, endOffset: Long): FetchResponseData.EpochEndOffset = {
       new FetchResponseData.EpochEndOffset()
@@ -143,7 +143,7 @@ class PartitionTest extends AbstractPartitionTest {
     ), leaderEpoch = 5)
     assertEquals(4, log.logEndOffset)
 
-    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true, log = log)
+    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true)
     assertEquals(Some(4), partition.leaderLogIfLocal.map(_.logEndOffset))
 
     val epochEndOffset = partition.lastOffsetForLeaderEpoch(currentLeaderEpoch = Optional.of[Integer](leaderEpoch),
@@ -171,7 +171,7 @@ class PartitionTest extends AbstractPartitionTest {
     ), leaderEpoch = 5)
     assertEquals(4, log.logEndOffset)
 
-    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true, log = log)
+    val partition = setupPartitionWithMocks(leaderEpoch = leaderEpoch, isLeader = true)
     assertEquals(Some(4), partition.leaderLogIfLocal.map(_.logEndOffset))
     assertEquals(None, log.latestEpoch)
 
@@ -733,8 +733,7 @@ class PartitionTest extends AbstractPartitionTest {
   }
 
   private def setupPartitionWithMocks(leaderEpoch: Int,
-                                      isLeader: Boolean,
-                                      log: Log = logManager.getOrCreateLog(topicPartition, topicId = None)): Partition = {
+                                      isLeader: Boolean): Partition = {
     partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None)
 
     val controllerEpoch = 0
@@ -2046,7 +2045,7 @@ class PartitionTest extends AbstractPartitionTest {
     leaderEpochCache,
     producerStateManager,
     logDirFailureChannel,
-    topicId = None,
+    _topicId = None,
     keepPartitionMetadataFile = true) {
 
     override def appendAsFollower(records: MemoryRecords): LogAppendInfo = {

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

@@ -121,7 +121,7 @@ class LogCleanerManagerTest extends Logging {
       extends Log(dir, config, segments, offsets.logStartOffset, offsets.recoveryPoint,
         offsets.nextOffsetMetadata, time.scheduler, new BrokerTopicStats, time,
         LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition, leaderEpochCache,
-        producerStateManager, logDirFailureChannel, topicId = None, keepPartitionMetadataFile = true) {
+        producerStateManager, logDirFailureChannel, _topicId = None, keepPartitionMetadataFile = true) {
       // Throw an error in getFirstBatchTimestampForSegments since it is called in grabFilthiestLog()
       override def getFirstBatchTimestampForSegments(segments: Iterable[LogSegment]): Iterable[Long] =
         throw new IllegalStateException("Error!")

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

@@ -136,7 +136,7 @@ class LogCleanerTest {
                       leaderEpochCache = leaderEpochCache,
                       producerStateManager = producerStateManager,
                       logDirFailureChannel = logDirFailureChannel,
-                      topicId = None,
+                      _topicId = None,
                       keepPartitionMetadataFile = true) {
       override def replaceSegments(newSegments: Seq[LogSegment], oldSegments: Seq[LogSegment], isRecoveredSwapFile: Boolean = false): Unit = {
         deleteStartLatch.countDown()

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

@@ -284,7 +284,7 @@ class LogLoaderTest {
       new Log(logDir, logConfig, interceptedLogSegments, offsets.logStartOffset, offsets.recoveryPoint,
         offsets.nextOffsetMetadata, mockTime.scheduler, brokerTopicStats, mockTime,
         LogManager.ProducerIdExpirationCheckIntervalMs, topicPartition, leaderEpochCache,
-        producerStateManager, logDirFailureChannel, topicId = None, keepPartitionMetadataFile = true)
+        producerStateManager, logDirFailureChannel, _topicId = None, keepPartitionMetadataFile = true)
     }
 
     // Retain snapshots for the last 2 segments
@@ -366,7 +366,7 @@ class LogLoaderTest {
       leaderEpochCache = leaderEpochCache,
       producerStateManager = stateManager,
       logDirFailureChannel = logDirFailureChannel,
-      topicId = None,
+      _topicId = None,
       keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
@@ -500,7 +500,7 @@ class LogLoaderTest {
       leaderEpochCache = leaderEpochCache,
       producerStateManager = stateManager,
       logDirFailureChannel = logDirFailureChannel,
-      topicId = None,
+      _topicId = None,
       keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
@@ -561,7 +561,7 @@ class LogLoaderTest {
       leaderEpochCache = leaderEpochCache,
       producerStateManager = stateManager,
       logDirFailureChannel = logDirFailureChannel,
-      topicId = None,
+      _topicId = None,
       keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)
@@ -624,7 +624,7 @@ class LogLoaderTest {
       leaderEpochCache = leaderEpochCache,
       producerStateManager = stateManager,
       logDirFailureChannel = logDirFailureChannel,
-      topicId = None,
+      _topicId = None,
       keepPartitionMetadataFile = true)
 
     EasyMock.verify(stateManager)

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

@@ -2319,9 +2319,8 @@ class LogTest {
     val log = createLog(logDir, logConfig)
 
     // Write a topic ID to the partition metadata file to ensure it is transferred correctly.
-    val id = Uuid.randomUuid()
-    log.topicId = Some(id)
-    log.partitionMetadataFile.write(id)
+    val topicId = Uuid.randomUuid()
+    log.assignTopicId(topicId)
 
     log.appendAsLeader(TestUtils.records(List(new SimpleRecord("foo".getBytes()))), leaderEpoch = 5)
     assertEquals(Some(5), log.latestEpoch)
@@ -2336,8 +2335,8 @@ class LogTest {
 
     // Check the topic ID remains in memory and was copied correctly.
     assertTrue(log.topicId.isDefined)
-    assertEquals(id, log.topicId.get)
-    assertEquals(id, log.partitionMetadataFile.read().topicId)
+    assertEquals(topicId, log.topicId.get)
+    assertEquals(topicId, log.partitionMetadataFile.read().topicId)
   }
 
   @Test

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

@@ -1075,7 +1075,6 @@ class ReplicaManagerTest {
   private def initializeLogAndTopicId(replicaManager: ReplicaManager, topicPartition: TopicPartition, topicId: Uuid): Unit = {
     val partition = replicaManager.createPartition(new TopicPartition(topic, 0))
     val log = replicaManager.logManager.getOrCreateLog(topicPartition, false, false, Some(topicId))
-    log.topicId = Some(topicId)
     partition.log = Some(log)
   }
 
@@ -1517,7 +1516,7 @@ class ReplicaManagerTest {
       leaderEpochCache = leaderEpochCache,
       producerStateManager = producerStateManager,
       logDirFailureChannel = mockLogDirFailureChannel,
-      topicId = topicId,
+      _topicId = topicId,
       keepPartitionMetadataFile = true) {
 
       override def endOffsetForEpoch(leaderEpoch: Int): Option[OffsetAndEpoch] = {

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

@@ -143,7 +143,7 @@ class SchedulerTest {
       recoveryPoint = offsets.recoveryPoint, nextOffsetMetadata = offsets.nextOffsetMetadata, scheduler,
       brokerTopicStats, mockTime, LogManager.ProducerIdExpirationCheckIntervalMs,
       topicPartition, leaderEpochCache, producerStateManager, logDirFailureChannel,
-      topicId = None, keepPartitionMetadataFile = true)
+      _topicId = None, keepPartitionMetadataFile = true)
     assertTrue(scheduler.taskRunning(log.producerExpireCheck))
     log.close()
     assertFalse(scheduler.taskRunning(log.producerExpireCheck))