Repository: kafka Updated Branches: refs/heads/0.10.2 e2d4198c4 -> 1ca2b1aac
KAFKA-5232; Fix Log.parseTopicPartitionName to handle deleted topics with a period in the name This issue would only be triggered if a broker was restarted while deletion was still taking place. Included a few minor improvements to that method and its tests. A separate PR was submitted for 0.10.2 as the trunk change did not merge cleanly. Author: Jaikiran Pai <[email protected]> Reviewers: Ryan Pridgeon <[email protected]>, Ismael Juma <[email protected]> Closes #3043 from jaikiran/KAFKA-5232 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/1ca2b1aa Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/1ca2b1aa Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/1ca2b1aa Branch: refs/heads/0.10.2 Commit: 1ca2b1aacc2994e441c1a949310129d3e78532b5 Parents: e2d4198 Author: Jaikiran Pai <[email protected]> Authored: Mon May 15 13:08:29 2017 +0100 Committer: Ismael Juma <[email protected]> Committed: Mon May 15 13:08:45 2017 +0100 ---------------------------------------------------------------------- core/src/main/scala/kafka/log/Log.scala | 40 +++++++--- core/src/main/scala/kafka/log/LogManager.scala | 11 +-- core/src/main/scala/kafka/log/LogSegment.scala | 2 +- .../src/test/scala/unit/kafka/log/LogTest.scala | 81 +++++++++++++++++--- 4 files changed, 103 insertions(+), 31 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/1ca2b1aa/core/src/main/scala/kafka/log/Log.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index 029e1ed..eba53bf 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -37,6 +37,7 @@ import com.yammer.metrics.core.Gauge import org.apache.kafka.common.utils.{Time, Utils} import kafka.message.{BrokerCompressionCodec, CompressionCodec, NoCompressionCodec} import org.apache.kafka.common.TopicPartition +import java.util.regex.Pattern object LogAppendInfo { val UnknownLogAppendInfo = LogAppendInfo(-1, -1, Record.NO_TIMESTAMP, -1L, Record.NO_TIMESTAMP, @@ -788,7 +789,7 @@ class Log(@volatile var dir: File, val start = time.nanoseconds lock synchronized { val newOffset = Math.max(expectedNextOffset, logEndOffset) - val logFile = logFilename(dir, newOffset) + val logFile = Log.logFile(dir, newOffset) val indexFile = indexFilename(dir, newOffset) val timeIndexFile = timeIndexFilename(dir, newOffset) for(file <- List(logFile, indexFile, timeIndexFile); if file.exists) { @@ -1080,6 +1081,8 @@ object Log { /** a directory that is scheduled to be deleted */ val DeleteDirSuffix = "-delete" + private val DeleteDirPattern = Pattern.compile(s"^(\\S+)-(\\S+)\\.(\\S+)$DeleteDirSuffix") + /** * Make log segment file name from offset bytes. All this does is pad out the offset number with zeros * so that ls sorts the files numerically. @@ -1101,9 +1104,18 @@ object Log { * @param dir The directory in which the log will reside * @param offset The base offset of the log file */ - def logFilename(dir: File, offset: Long) = + def logFile(dir: File, offset: Long) = new File(dir, filenamePrefixFromOffset(offset) + LogFileSuffix) + /** + * Return a directory name to rename the log directory to for async deletion. The name will be in the following + * format: topic-partition.uniqueId-delete where topic, partition and uniqueId are variables. + */ + def logDeleteDirName(logName: String): String = { + val uniqueId = java.util.UUID.randomUUID.toString.replaceAll("-", "") + s"$logName.$uniqueId$DeleteDirSuffix" + } + /** * Construct an index file name in the given dir using the given base offset * @@ -1126,32 +1138,36 @@ object Log { * Parse the topic and partition out of the directory name of a log */ def parseTopicPartitionName(dir: File): TopicPartition = { + if (dir == null) + throw new KafkaException("dir should not be null") def exception(dir: File): KafkaException = { - new KafkaException("Found directory " + dir.getCanonicalPath + ", " + - "'" + dir.getName + "' is not in the form of topic-partition or " + - "ongoing-deleting directory(topic-partition.uniqueId-delete)\n" + - "If a directory does not contain Kafka topic data it should not exist in Kafka's log " + - "directory") + new KafkaException(s"Found directory ${dir.getCanonicalPath}, '${dir.getName}' is not in the form of " + + "topic-partition or topic-partition.uniqueId-delete (if marked for deletion).\n" + + "Kafka's log directories (and children) should only contain Kafka topic data.") } val dirName = dir.getName if (dirName == null || dirName.isEmpty || !dirName.contains('-')) throw exception(dir) - if (dirName.endsWith(DeleteDirSuffix) && !dirName.matches("^(\\S+)-(\\S+)\\.(\\S+)" + DeleteDirSuffix)) + if (dirName.endsWith(DeleteDirSuffix) && !DeleteDirPattern.matcher(dirName).matches) throw exception(dir) val name: String = - if (dirName.endsWith(DeleteDirSuffix)) dirName.substring(0, dirName.indexOf('.')) + if (dirName.endsWith(DeleteDirSuffix)) dirName.substring(0, dirName.lastIndexOf('.')) else dirName val index = name.lastIndexOf('-') val topic = name.substring(0, index) - val partition = name.substring(index + 1) - if (topic.length < 1 || partition.length < 1) + val partitionString = name.substring(index + 1) + if (topic.isEmpty || partitionString.isEmpty) throw exception(dir) - new TopicPartition(topic, partition.toInt) + val partition = + try partitionString.toInt + catch { case _: NumberFormatException => throw exception(dir) } + + new TopicPartition(topic, partition) } } http://git-wip-us.apache.org/repos/asf/kafka/blob/1ca2b1aa/core/src/main/scala/kafka/log/LogManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index 761edf9..8455aa0 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -407,20 +407,15 @@ class LogManager(val logDirs: Array[File], */ def asyncDelete(topicPartition: TopicPartition) = { val removedLog: Log = logCreationOrDeletionLock synchronized { - logs.remove(topicPartition) - } + logs.remove(topicPartition) + } if (removedLog != null) { //We need to wait until there is no more cleaning task on the log to be deleted before actually deleting it. if (cleaner != null) { cleaner.abortCleaning(topicPartition) cleaner.updateCheckpoints(removedLog.dir.getParentFile) } - // renaming the directory to topic-partition.uniqueId-delete - val dirName = new StringBuilder(removedLog.name) - .append(".") - .append(java.util.UUID.randomUUID.toString.replaceAll("-","")) - .append(Log.DeleteDirSuffix) - .toString() + val dirName = Log.logDeleteDirName(removedLog.name) removedLog.close() val renamedDir = new File(removedLog.dir.getParent, dirName) val renameSuccessful = removedLog.dir.renameTo(renamedDir) http://git-wip-us.apache.org/repos/asf/kafka/blob/1ca2b1aa/core/src/main/scala/kafka/log/LogSegment.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/log/LogSegment.scala b/core/src/main/scala/kafka/log/LogSegment.scala index 8854c3a..9da56c1 100755 --- a/core/src/main/scala/kafka/log/LogSegment.scala +++ b/core/src/main/scala/kafka/log/LogSegment.scala @@ -68,7 +68,7 @@ class LogSegment(val log: FileRecords, @volatile private var offsetOfMaxTimestamp = timeIndex.lastEntry.offset def this(dir: File, startOffset: Long, indexIntervalBytes: Int, maxIndexSize: Int, rollJitterMs: Long, time: Time, fileAlreadyExists: Boolean = false, initFileSize: Int = 0, preallocate: Boolean = false) = - this(FileRecords.open(Log.logFilename(dir, startOffset), fileAlreadyExists, initFileSize, preallocate), + this(FileRecords.open(Log.logFile(dir, startOffset), fileAlreadyExists, initFileSize, preallocate), new OffsetIndex(Log.indexFilename(dir, startOffset), baseOffset = startOffset, maxIndexSize = maxIndexSize), new TimeIndex(Log.timeIndexFilename(dir, startOffset), baseOffset = startOffset, maxIndexSize = maxIndexSize), startOffset, http://git-wip-us.apache.org/repos/asf/kafka/blob/1ca2b1aa/core/src/test/scala/unit/kafka/log/LogTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index cc46fe6..b4dff6d 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogTest.scala @@ -54,7 +54,7 @@ class LogTest extends JUnitSuite { def createEmptyLogs(dir: File, offsets: Int*) { for(offset <- offsets) { - Log.logFilename(dir, offset).createNewFile() + Log.logFile(dir, offset).createNewFile() Log.indexFilename(dir, offset).createNewFile() } } @@ -1095,12 +1095,27 @@ class LogTest extends JUnitSuite { def testParseTopicPartitionName() { val topic = "test_topic" val partition = "143" - val dir = new File(logDir + topicPartitionName(topic, partition)) + val dir = new File(logDir, topicPartitionName(topic, partition)) val topicPartition = Log.parseTopicPartitionName(dir) assertEquals(topic, topicPartition.topic) assertEquals(partition.toInt, topicPartition.partition) } + /** + * Tests that log directories with a period in their name that have been marked for deletion + * are parsed correctly by `Log.parseTopicPartitionName` (see KAFKA-5232 for details). + */ + @Test + def testParseTopicPartitionNameWithPeriodForDeletedTopic() { + val topic = "foo.bar-testtopic" + val partition = "42" + val dir = new File(logDir, Log.logDeleteDirName(topicPartitionName(topic, partition))) + val topicPartition = Log.parseTopicPartitionName(dir) + assertEquals("Unexpected topic name parsed", topic, topicPartition.topic) + assertEquals("Unexpected partition number parsed", partition.toInt, topicPartition.partition) + } + + @Test def testParseTopicPartitionNameForEmptyName() { try { @@ -1108,7 +1123,7 @@ class LogTest extends JUnitSuite { Log.parseTopicPartitionName(dir) fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath) } catch { - case _: Exception => // its GOOD! + case _: KafkaException => // expected } } @@ -1119,7 +1134,7 @@ class LogTest extends JUnitSuite { Log.parseTopicPartitionName(dir) fail("KafkaException should have been thrown for dir: " + dir) } catch { - case _: Exception => // its GOOD! + case _: KafkaException => // expected } } @@ -1127,12 +1142,20 @@ class LogTest extends JUnitSuite { def testParseTopicPartitionNameForMissingSeparator() { val topic = "test_topic" val partition = "1999" - val dir = new File(logDir + File.separator + topic + partition) + val dir = new File(logDir, topic + partition) try { Log.parseTopicPartitionName(dir) fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath) } catch { - case _: Exception => // its GOOD! + case _: KafkaException => // expected + } + // also test the "-delete" marker case + val deleteMarkerDir = new File(logDir, Log.logDeleteDirName(topic + partition)) + try { + Log.parseTopicPartitionName(deleteMarkerDir) + fail("KafkaException should have been thrown for dir: " + deleteMarkerDir.getCanonicalPath) + } catch { + case _: KafkaException => // expected } } @@ -1140,13 +1163,22 @@ class LogTest extends JUnitSuite { def testParseTopicPartitionNameForMissingTopic() { val topic = "" val partition = "1999" - val dir = new File(logDir + topicPartitionName(topic, partition)) + val dir = new File(logDir, topicPartitionName(topic, partition)) try { Log.parseTopicPartitionName(dir) fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath) } catch { - case _: Exception => // its GOOD! + case _: KafkaException => // expected } + // also test the "-delete" marker case + val deleteMarkerDir = new File(logDir, Log.logDeleteDirName(topicPartitionName(topic, partition))) + try { + Log.parseTopicPartitionName(deleteMarkerDir) + fail("KafkaException should have been thrown for dir: " + deleteMarkerDir.getCanonicalPath) + } catch { + case _: KafkaException => // expected + } + } @Test @@ -1158,7 +1190,36 @@ class LogTest extends JUnitSuite { Log.parseTopicPartitionName(dir) fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath) } catch { - case _: Exception => // its GOOD! + case _: KafkaException => // its GOOD! + } + // also test the "-delete" marker case + val deleteMarkerDir = new File(logDir, Log.logDeleteDirName(topicPartitionName(topic, partition))) + try { + Log.parseTopicPartitionName(deleteMarkerDir) + fail("KafkaException should have been thrown for dir: " + deleteMarkerDir.getCanonicalPath) + } catch { + case _: KafkaException => // expected + } + } + + @Test + def testParseTopicPartitionNameForInvalidPartition() { + val topic = "test_topic" + val partition = "1999a" + val dir = new File(logDir, topicPartitionName(topic, partition)) + try { + Log.parseTopicPartitionName(dir) + fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath) + } catch { + case _: KafkaException => // expected + } + // also test the "-delete" marker case + val deleteMarkerDir = new File(logDir, Log.logDeleteDirName(topic + partition)) + try { + Log.parseTopicPartitionName(deleteMarkerDir) + fail("KafkaException should have been thrown for dir: " + deleteMarkerDir.getCanonicalPath) + } catch { + case _: KafkaException => // expected } } @@ -1181,7 +1242,7 @@ class LogTest extends JUnitSuite { } def topicPartitionName(topic: String, partition: String): String = - File.separator + topic + "-" + partition + topic + "-" + partition @Test def testDeleteOldSegmentsMethod() {
