diff --git a/core/src/main/scala/kafka/log/AbstractIndex.scala b/core/src/main/scala/kafka/log/AbstractIndex.scala index 05237c44193fc..bae5d84d0fceb 100644 --- a/core/src/main/scala/kafka/log/AbstractIndex.scala +++ b/core/src/main/scala/kafka/log/AbstractIndex.scala @@ -175,7 +175,8 @@ abstract class AbstractIndex[K, V](@volatile var file: File, val baseOffset: Lon val roundedNewSize = roundDownToExactMultiple(newSize, entrySize) if (_length == roundedNewSize) { - debug(s"Index ${file.getAbsolutePath} was not resized because it already has size $roundedNewSize") + if (isDebugEnabled) + debug(s"Index ${file.getAbsolutePath} was not resized because it already has size $roundedNewSize") false } else { val raf = new RandomAccessFile(file, "rw") @@ -190,8 +191,9 @@ abstract class AbstractIndex[K, V](@volatile var file: File, val baseOffset: Lon mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, roundedNewSize) _maxEntries = mmap.limit() / entrySize mmap.position(position) - debug(s"Resized ${file.getAbsolutePath} to $roundedNewSize, position is ${mmap.position()} " + - s"and limit is ${mmap.limit()}") + if (isDebugEnabled) + debug(s"Resized ${file.getAbsolutePath} to $roundedNewSize, position is ${mmap.position()} " + + s"and limit is ${mmap.limit()}") true } finally { CoreUtils.swallow(raf.close(), AbstractIndex) diff --git a/core/src/main/scala/kafka/log/LogSegment.scala b/core/src/main/scala/kafka/log/LogSegment.scala index f25813903225d..1b60ac0af5bf8 100755 --- a/core/src/main/scala/kafka/log/LogSegment.scala +++ b/core/src/main/scala/kafka/log/LogSegment.scala @@ -43,8 +43,8 @@ import scala.math._ * A segment with a base offset of [base_offset] would be stored in two files, a [base_offset].index and a [base_offset].log file. * * @param log The file records containing log entries - * @param offsetIndex The offset index - * @param timeIndex The timestamp index + * @param lazyOffsetIndex The offset index + * @param lazyTimeIndex The timestamp index * @param txnIndex The transaction index * @param baseOffset A lower bound on the offsets in this segment * @param indexIntervalBytes The approximate number of bytes between entries in the index diff --git a/core/src/main/scala/kafka/log/OffsetIndex.scala b/core/src/main/scala/kafka/log/OffsetIndex.scala index d043d0a50271a..93a1c036b39d3 100755 --- a/core/src/main/scala/kafka/log/OffsetIndex.scala +++ b/core/src/main/scala/kafka/log/OffsetIndex.scala @@ -59,8 +59,9 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl /* the last offset in the index */ private[this] var _lastOffset = lastEntry.offset - debug(s"Loaded index file ${file.getAbsolutePath} with maxEntries = $maxEntries, " + - s"maxIndexSize = $maxIndexSize, entries = ${_entries}, lastOffset = ${_lastOffset}, file position = ${mmap.position()}") + if (isDebugEnabled) + debug(s"Loaded index file ${file.getAbsolutePath} with maxEntries = $maxEntries, " + + s"maxIndexSize = $maxIndexSize, entries = ${_entries}, lastOffset = ${_lastOffset}, file position = ${mmap.position()}") /** * The last entry in the index @@ -136,13 +137,14 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl /** * Append an entry for the given offset/location pair to the index. This entry must have a larger offset than all subsequent entries. - * @throws IndexOffsetOverflowException if the offset causes index offset to overflow + * @throws kafka.common.IndexOffsetOverflowException#IndexOffsetOverflowException if the offset causes index offset to overflow */ def append(offset: Long, position: Int) { inLock(lock) { require(!isFull, "Attempt to append to a full index (size = " + _entries + ").") if (_entries == 0 || offset > _lastOffset) { - trace(s"Adding index entry $offset => $position to ${file.getAbsolutePath}") + if (isTraceEnabled) + trace(s"Adding index entry $offset => $position to ${file.getAbsolutePath}") mmap.putInt(relativeOffset(offset)) mmap.putInt(position) _entries += 1 @@ -186,8 +188,9 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl _entries = entries mmap.position(_entries * entrySize) _lastOffset = lastEntry.offset - debug(s"Truncated index ${file.getAbsolutePath} to $entries entries;" + - s" position is now ${mmap.position()} and last offset is now ${_lastOffset}") + if (isDebugEnabled) + debug(s"Truncated index ${file.getAbsolutePath} to $entries entries;" + + s" position is now ${mmap.position()} and last offset is now ${_lastOffset}") } } diff --git a/core/src/main/scala/kafka/log/TimeIndex.scala b/core/src/main/scala/kafka/log/TimeIndex.scala index f9bc929178ad4..b2bafeb82fc02 100644 --- a/core/src/main/scala/kafka/log/TimeIndex.scala +++ b/core/src/main/scala/kafka/log/TimeIndex.scala @@ -58,8 +58,9 @@ class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: override def entrySize = 12 - debug(s"Loaded index file ${file.getAbsolutePath} with maxEntries = $maxEntries, maxIndexSize = $maxIndexSize," + - s" entries = ${_entries}, lastOffset = ${_lastEntry}, file position = ${mmap.position()}") + if (isDebugEnabled) + debug(s"Loaded index file ${file.getAbsolutePath} with maxEntries = $maxEntries, maxIndexSize = $maxIndexSize," + + s" entries = ${_entries}, lastOffset = ${_lastEntry}, file position = ${mmap.position()}") // We override the full check to reserve the last time index entry slot for the on roll call. override def isFull: Boolean = entries >= maxEntries - 1 @@ -120,17 +121,20 @@ class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: // because that could happen in the following two scenarios: // 1. A log segment is closed. // 2. LogSegment.onBecomeInactiveSegment() is called when an active log segment is rolled. - if (_entries != 0 && offset < lastEntry.offset) - throw new InvalidOffsetException(s"Attempt to append an offset ($offset) to slot ${_entries} no larger than" + - s" the last offset appended (${lastEntry.offset}) to ${file.getAbsolutePath}.") - if (_entries != 0 && timestamp < lastEntry.timestamp) - throw new IllegalStateException(s"Attempt to append a timestamp ($timestamp) to slot ${_entries} no larger" + - s" than the last timestamp appended (${lastEntry.timestamp}) to ${file.getAbsolutePath}.") + if (_entries != 0) { + if (offset < lastEntry.offset) + throw new InvalidOffsetException(s"Attempt to append an offset ($offset) to slot ${_entries} no larger than" + + s" the last offset appended (${lastEntry.offset}) to ${file.getAbsolutePath}.") + if (timestamp < lastEntry.timestamp) + throw new IllegalStateException(s"Attempt to append a timestamp ($timestamp) to slot ${_entries} no larger" + + s" than the last timestamp appended (${lastEntry.timestamp}) to ${file.getAbsolutePath}.") + } // We only append to the time index when the timestamp is greater than the last inserted timestamp. // If all the messages are in message format v0, the timestamp will always be NoTimestamp. In that case, the time // index will be empty. if (timestamp > lastEntry.timestamp) { - trace(s"Adding index entry $timestamp => $offset to ${file.getAbsolutePath}.") + if (isTraceEnabled) + trace(s"Adding index entry $timestamp => $offset to ${file.getAbsolutePath}.") mmap.putLong(timestamp) mmap.putInt(relativeOffset(offset)) _entries += 1 @@ -204,7 +208,8 @@ class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: _entries = entries mmap.position(_entries * entrySize) _lastEntry = lastEntryFromIndexFile - debug(s"Truncated index ${file.getAbsolutePath} to $entries entries; position is now ${mmap.position()} and last entry is now ${_lastEntry}") + if (isDebugEnabled) + debug(s"Truncated index ${file.getAbsolutePath} to $entries entries; position is now ${mmap.position()} and last entry is now ${_lastEntry}") } }