diff --git a/core/src/main/scala/kafka/log/LazyIndex.scala b/core/src/main/scala/kafka/log/LazyIndex.scala index cd7e0e5f61ca7..d83fc581af49d 100644 --- a/core/src/main/scala/kafka/log/LazyIndex.scala +++ b/core/src/main/scala/kafka/log/LazyIndex.scala @@ -17,27 +17,32 @@ package kafka.log -import java.io.File +import java.io.{File, IOException} +import java.nio.file.Files import java.util.concurrent.locks.ReentrantLock import LazyIndex._ import kafka.utils.CoreUtils.inLock import kafka.utils.threadsafe +import org.apache.kafka.common.utils.Utils /** * A wrapper over an `AbstractIndex` instance that provides a mechanism to defer loading (i.e. memory mapping) the * underlying index until it is accessed for the first time via the `get` method. * - * This is an important optimization with regards to broker start-up time if it has a large number of segments. + * This is an important optimization with regards to broker start-up and shutdown time if it has a large number of segments. + * It prevents illegal accesses to the underlying index after closing the index, which might otherwise lead to memory + * leaks due to recreation of underlying memory mapped object. * - * Methods of this class are thread safe. Make sure to check `AbstractIndex` subclasses documentation - * to establish their thread safety. + * Finally, this wrapper ensures that redundant disk accesses and memory mapped operations are avoided upon attempts to + * delete or rename the file that backs this index. * * @param loadIndex A function that takes a `File` pointing to an index and returns a loaded `AbstractIndex` instance. */ @threadsafe class LazyIndex[T <: AbstractIndex] private (@volatile private var indexWrapper: IndexWrapper, loadIndex: File => T) { - + // A closed index does not allow accessing its indices to prevent side effects. + @volatile private var isClosed: Boolean = false private val lock = new ReentrantLock() def file: File = indexWrapper.file @@ -49,6 +54,8 @@ class LazyIndex[T <: AbstractIndex] private (@volatile private var indexWrapper: } def get: T = { + if (isClosed) + throw new IllegalStateException(s"Attempt to access the closed Index (file=$file).") indexWrapper match { case indexValue: IndexValue[T] => indexValue.index case _: IndexFile => @@ -64,6 +71,63 @@ class LazyIndex[T <: AbstractIndex] private (@volatile private var indexWrapper: } } + /** + * Close this index file. + * Note: This will be a no-op if the index has already been closed. + */ + def close(): Unit = { + if (!isClosed) { + inLock(lock) { + indexWrapper match { + case indexValue: IndexValue[T] => indexValue.index.close() + case _: IndexFile => // no-op + } + } + isClosed = true + } + } + + /** + * Delete the index file that backs this index if exists. + * This method ensures that if the index file has already been closed or in case it has not been created before, + * it will not be recreated as a side effect. + * + * @throws IOException if deletion fails due to an I/O error + * @return `true` if the file was deleted by this method; `false` if the file could not be deleted because it did + * not exist + */ + def deleteIfExists(): Boolean = { + if (isClosed) + Files.deleteIfExists(file.toPath) + else { + inLock(lock) { + indexWrapper match { + case indexValue: IndexValue[T] => indexValue.index.deleteIfExists() + case _: IndexFile => Files.deleteIfExists(file.toPath) + } + } + } + } + + /** + * Rename the file that backs this index if the index has ever been initialized or file already exists. + * + * @throws IOException if rename fails for defined index or existing file. + */ + def renameTo(f: File) { + try { + if (file.exists) + Utils.atomicMoveWithFallback(file.toPath, f.toPath) + else { + inLock(lock) { + indexWrapper match { + case indexValue: IndexValue[T] => Utils.atomicMoveWithFallback(file.toPath, f.toPath) + case _: IndexFile => // no-op + } + } + } + } finally file = f + } } object LazyIndex { diff --git a/core/src/main/scala/kafka/log/LogSegment.scala b/core/src/main/scala/kafka/log/LogSegment.scala index a54c2c2eacc74..70e2ea6693d05 100755 --- a/core/src/main/scala/kafka/log/LogSegment.scala +++ b/core/src/main/scala/kafka/log/LogSegment.scala @@ -488,13 +488,13 @@ class LogSegment private[log] (val log: FileRecords, } /** - * Change the suffix for the index and log file for this log segment + * Change the suffix for the index and log files for this log segment * IOException from this method should be handled by the caller */ def changeFileSuffixes(oldSuffix: String, newSuffix: String): Unit = { log.renameTo(new File(CoreUtils.replaceSuffix(log.file.getPath, oldSuffix, newSuffix))) - offsetIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyOffsetIndex.file.getPath, oldSuffix, newSuffix))) - timeIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyTimeIndex.file.getPath, oldSuffix, newSuffix))) + lazyOffsetIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyOffsetIndex.file.getPath, oldSuffix, newSuffix))) + lazyTimeIndex.renameTo(new File(CoreUtils.replaceSuffix(lazyTimeIndex.file.getPath, oldSuffix, newSuffix))) txnIndex.renameTo(new File(CoreUtils.replaceSuffix(txnIndex.file.getPath, oldSuffix, newSuffix))) } @@ -584,10 +584,12 @@ class LogSegment private[log] (val log: FileRecords, /** * Close this log segment */ - def close(): Unit = { - CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true), this) - CoreUtils.swallow(offsetIndex.close(), this) - CoreUtils.swallow(timeIndex.close(), this) + def close() { + if (_maxTimestampSoFar.nonEmpty || _offsetOfMaxTimestampSoFar.nonEmpty) { + CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true), this) + } + CoreUtils.swallow(lazyOffsetIndex.close(), this) + CoreUtils.swallow(lazyTimeIndex.close(), this) CoreUtils.swallow(log.close(), this) CoreUtils.swallow(txnIndex.close(), this) } @@ -595,9 +597,9 @@ class LogSegment private[log] (val log: FileRecords, /** * Close file handlers used by the log segment but don't write to disk. This is used when the disk may have failed */ - def closeHandlers(): Unit = { - CoreUtils.swallow(offsetIndex.closeHandler(), this) - CoreUtils.swallow(timeIndex.closeHandler(), this) + def closeHandlers() { + CoreUtils.swallow(lazyOffsetIndex.close(), this) + CoreUtils.swallow(lazyTimeIndex.close(), this) CoreUtils.swallow(log.closeHandlers(), this) CoreUtils.swallow(txnIndex.close(), this) } @@ -620,8 +622,8 @@ class LogSegment private[log] (val log: FileRecords, CoreUtils.tryAll(Seq( () => delete(log.deleteIfExists _, "log", log.file, logIfMissing = true), - () => delete(offsetIndex.deleteIfExists _, "offset index", lazyOffsetIndex.file, logIfMissing = true), - () => delete(timeIndex.deleteIfExists _, "time index", lazyTimeIndex.file, logIfMissing = true), + () => delete(lazyOffsetIndex.deleteIfExists _, "offset index", lazyOffsetIndex.file, logIfMissing = true), + () => delete(lazyTimeIndex.deleteIfExists _, "time index", lazyTimeIndex.file, logIfMissing = true), () => delete(txnIndex.deleteIfExists _, "transaction index", txnIndex.file, logIfMissing = false) )) } diff --git a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala index 5e0a1a83c3af0..64045d52cd832 100644 --- a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala @@ -24,7 +24,7 @@ import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.record._ import org.apache.kafka.common.utils.{MockTime, Time, Utils} import org.junit.Assert._ -import org.junit.{After, Before, Test} +import org.junit.{After, Assert, Before, Test} import scala.collection.JavaConverters._ import scala.collection._ @@ -136,6 +136,35 @@ class LogSegmentTest { } } + /** + * This tests the scenario, where the caller attempts to access the offsets of a closed segment. + * + * Accessing underlying indices of a closed segment would lead to memory leaks due to recreation of the underlying + * memory mapped objects. + */ + @Test + def testIndexAccessAfterClosingSegment() { + val seg = createSegment(0, time = new MockTime) + // Accessing the indices while the segment is legal. + seg.offsetIndex + seg.timeIndex + + // Accessing the indices while the segment is closed is disallowed. + seg.close() + try { + seg.offsetIndex + Assert.fail("Expected IllegalStateException due to accessing OffsetIndex of a closed segment.") + } catch { + case _: IllegalStateException => //expected + } + try { + seg.timeIndex + Assert.fail("Expected IllegalStateException due to accessing TimeIndex of a closed segment.") + } catch { + case _: IllegalStateException => //expected + } + } + @Test def testTruncateEmptySegment(): Unit = { // This tests the scenario in which the follower truncates to an empty segment. In this @@ -145,11 +174,13 @@ class LogSegmentTest { val maxSegmentMs = 300000 val time = new MockTime val seg = createSegment(0, time = time) + val timeIndexSizeInBytes = seg.timeIndex.sizeInBytes + val offsetIndexSizeInBytes = seg.offsetIndex.sizeInBytes seg.close() val reopened = createSegment(0, time = time) - assertEquals(0, seg.timeIndex.sizeInBytes) - assertEquals(0, seg.offsetIndex.sizeInBytes) + assertEquals(0, timeIndexSizeInBytes) + assertEquals(0, offsetIndexSizeInBytes) time.sleep(500) reopened.truncateTo(57) @@ -262,11 +293,24 @@ class LogSegmentTest { val seg = createSegment(40) val logFile = seg.log.file val indexFile = seg.lazyOffsetIndex.file + val timeIndexFile = seg.lazyTimeIndex.file + // Ensure that files for offset and time indices have not been created redundantly. + assertFalse(seg.lazyOffsetIndex.file.exists) + assertFalse(seg.lazyTimeIndex.file.exists) seg.changeFileSuffixes("", ".deleted") + // Ensure that attempt to change suffixes for non-existing offset and time indices do not create new files. + assertFalse(seg.lazyOffsetIndex.file.exists) + assertFalse(seg.lazyTimeIndex.file.exists) + // Ensure that file names are updated accordingly. assertEquals(logFile.getAbsolutePath + ".deleted", seg.log.file.getAbsolutePath) assertEquals(indexFile.getAbsolutePath + ".deleted", seg.lazyOffsetIndex.file.getAbsolutePath) assertTrue(seg.log.file.exists) + // Ensure lazy creation of offset index file upon accessing it. + seg.lazyOffsetIndex.get assertTrue(seg.lazyOffsetIndex.file.exists) + // Ensure lazy creation of time index file upon accessing it. + seg.lazyTimeIndex.get + assertTrue(seg.lazyTimeIndex.file.exists) } /**