From 5ab2370e03f99792d809028abbc024e44a50626a Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 25 May 2021 06:39:15 -0700 Subject: [PATCH 1/3] Don't ignore delete of partition metadata file --- core/src/main/scala/kafka/log/Log.scala | 6 +++++- .../src/main/scala/kafka/server/PartitionMetadataFile.scala | 4 ++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 0a4851ba6842a..34cae1f25b6b4 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -325,7 +325,11 @@ class Log(@volatile private var _dir: File, // Ensure we do not try to assign a provided topicId that is inconsistent with the ID on file. if (partitionMetadataFile.exists()) { if (!keepPartitionMetadataFile) - partitionMetadataFile.delete() + try partitionMetadataFile.delete() + catch { + case e: IOException => + error(s"Error while trying to delete partition metadata file ${partitionMetadataFile}", e) + } else { val fileTopicId = partitionMetadataFile.read().topicId if (topicId.isDefined && !topicId.contains(fileTopicId)) diff --git a/core/src/main/scala/kafka/server/PartitionMetadataFile.scala b/core/src/main/scala/kafka/server/PartitionMetadataFile.scala index 25b1ba6129d2b..65f2bc059d8d8 100644 --- a/core/src/main/scala/kafka/server/PartitionMetadataFile.scala +++ b/core/src/main/scala/kafka/server/PartitionMetadataFile.scala @@ -138,7 +138,7 @@ class PartitionMetadataFile(val file: File, file.exists() } - def delete(): Boolean = { - file.delete() + def delete(): Unit = { + Files.delete(file.toPath) } } From ff34e36d763c0031eeda92ae9faf2aa4dbb12585 Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 25 May 2021 06:39:57 -0700 Subject: [PATCH 2/3] Don't expose log topic id mutability outside of assignTopicId --- core/src/main/scala/kafka/log/Log.scala | 12 +++++++----- .../scala/unit/kafka/cluster/PartitionLockTest.scala | 2 +- .../scala/unit/kafka/cluster/PartitionTest.scala | 11 +++++------ .../scala/unit/kafka/log/LogCleanerManagerTest.scala | 2 +- .../test/scala/unit/kafka/log/LogCleanerTest.scala | 2 +- .../test/scala/unit/kafka/log/LogLoaderTest.scala | 10 +++++----- core/src/test/scala/unit/kafka/log/LogTest.scala | 9 ++++----- .../scala/unit/kafka/server/ReplicaManagerTest.scala | 3 +-- .../test/scala/unit/kafka/utils/SchedulerTest.scala | 2 +- 9 files changed, 26 insertions(+), 27 deletions(-) diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 34cae1f25b6b4..f07aa1adf5de2 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._ @@ -332,16 +332,18 @@ class Log(@volatile private var _dir: File, } else { 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 if (keepPartitionMetadataFile) { - topicId.foreach(partitionMetadataFile.write) + _topicId.foreach(partitionMetadataFile.write) } } + def topicId: Option[Uuid] = _topicId + def dir: File = _dir def parentDir: String = _parentDir @@ -557,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/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 fbdbe207a1898..a11cc6c81878f 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 c137eab0a4194..e2d4e0d7d3a77 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 d59ed1d0c4c63..2e2b1b12f70a4 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 d1d38601bda93..451f1c59498de 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 93362e25a9d8c..494a79d39c112 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)) From fdb1a5fd999e3bff683993194da73b0fff400cd7 Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Thu, 27 May 2021 06:43:34 -0700 Subject: [PATCH 3/3] Implement PartitionMetadataFile.toString --- core/src/main/scala/kafka/server/PartitionMetadataFile.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/core/src/main/scala/kafka/server/PartitionMetadataFile.scala b/core/src/main/scala/kafka/server/PartitionMetadataFile.scala index 65f2bc059d8d8..c08c87dabe492 100644 --- a/core/src/main/scala/kafka/server/PartitionMetadataFile.scala +++ b/core/src/main/scala/kafka/server/PartitionMetadataFile.scala @@ -141,4 +141,6 @@ class PartitionMetadataFile(val file: File, def delete(): Unit = { Files.delete(file.toPath) } + + override def toString: String = s"PartitionMetadataFile(path=$path)" }