diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 82e083f0c766a..2862851e37163 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/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 { diff --git a/core/src/main/scala/kafka/server/PartitionMetadataFile.scala b/core/src/main/scala/kafka/server/PartitionMetadataFile.scala index 25b1ba6129d2b..c08c87dabe492 100644 --- a/core/src/main/scala/kafka/server/PartitionMetadataFile.scala +++ b/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)" } diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala index 0cbeec9798c13..8d567ec91696a 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala +++ b/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 = { diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala index 703071568a0c0..b614e13d76e32 100644 --- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala +++ b/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 = { diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala index 7620348264711..19de8ea2703bb 100644 --- a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala +++ b/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!") diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala index 0890b038c853b..5c91041b9f942 100755 --- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala +++ b/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() diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala index 4826784836303..9a7b627492472 100644 --- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala +++ b/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) diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index 30541165a8289..4cc88d0f5873c 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/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 diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 28f7be88c0426..99928f0395920 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/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] = { diff --git a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala index e5d2f2176c393..b085ab0166513 100644 --- a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala +++ b/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))