This is an automated email from the ASF dual-hosted git repository. jgus pushed a commit to branch 1.1 in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/1.1 by this push: new daebb01 MINOR: Improve logging around index files (#6385) daebb01 is described below commit daebb01622c6190da00f7a7e37be7d187989ac34 Author: Bob Barrett <bob.barr...@outlook.com> AuthorDate: Tue Mar 19 02:00:01 2019 -0400 MINOR: Improve logging around index files (#6385) This patch adds additional DEBUG statements in AbstractIndex.scala, OffsetIndex.scala, and TimeIndex.scala. It also changes the logging on append from DEBUG to TRACE to make DEBUG logging less disruptive, and it ensures that exceptions raised from index classes include file/offset information. Reviewers: Jason Gustafson <ja...@confluent.io> --- core/src/main/scala/kafka/log/AbstractIndex.scala | 3 +++ core/src/main/scala/kafka/log/OffsetIndex.scala | 15 +++++++++------ core/src/main/scala/kafka/log/TimeIndex.scala | 17 +++++++++++------ core/src/main/scala/kafka/log/TransactionIndex.scala | 13 +++++++------ 4 files changed, 30 insertions(+), 18 deletions(-) diff --git a/core/src/main/scala/kafka/log/AbstractIndex.scala b/core/src/main/scala/kafka/log/AbstractIndex.scala index 44083c1..ec91b4e 100644 --- a/core/src/main/scala/kafka/log/AbstractIndex.scala +++ b/core/src/main/scala/kafka/log/AbstractIndex.scala @@ -115,6 +115,7 @@ 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") false } else { val raf = new RandomAccessFile(file, "rw") @@ -129,6 +130,8 @@ 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()}") true } finally { CoreUtils.swallow(raf.close(), this) diff --git a/core/src/main/scala/kafka/log/OffsetIndex.scala b/core/src/main/scala/kafka/log/OffsetIndex.scala index 523c88c..400388b 100755 --- a/core/src/main/scala/kafka/log/OffsetIndex.scala +++ b/core/src/main/scala/kafka/log/OffsetIndex.scala @@ -57,8 +57,8 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl /* the last offset in the index */ private[this] var _lastOffset = lastEntry.offset - debug("Loaded index file %s with maxEntries = %d, maxIndexSize = %d, entries = %d, lastOffset = %d, file position = %d" - .format(file.getAbsolutePath, maxEntries, maxIndexSize, _entries, _lastOffset, mmap.position())) + 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 @@ -126,7 +126,8 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl def entry(n: Int): OffsetPosition = { maybeLock(lock) { if(n >= _entries) - throw new IllegalArgumentException("Attempt to fetch the %dth entry from an index of size %d.".format(n, _entries)) + throw new IllegalArgumentException(s"Attempt to fetch the ${n}th entry from index ${file.getAbsolutePath}, " + + s"which has size ${_entries}.") val idx = mmap.duplicate OffsetPosition(relativeOffset(idx, n), physical(idx, n)) } @@ -139,15 +140,15 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl inLock(lock) { require(!isFull, "Attempt to append to a full index (size = " + _entries + ").") if (_entries == 0 || offset > _lastOffset) { - debug("Adding index entry %d => %d to %s.".format(offset, position, file.getName)) + trace(s"Adding index entry $offset => $position to ${file.getAbsolutePath}") mmap.putInt((offset - baseOffset).toInt) mmap.putInt(position) _entries += 1 _lastOffset = offset require(_entries * entrySize == mmap.position(), entries + " entries but file position in index is " + mmap.position() + ".") } else { - throw new InvalidOffsetException("Attempt to append an offset (%d) to position %d no larger than the last offset appended (%d) to %s." - .format(offset, entries, _lastOffset, file.getAbsolutePath)) + throw new InvalidOffsetException(s"Attempt to append an offset ($offset) to position $entries no larger than" + + s" the last offset appended (${_lastOffset}) to ${file.getAbsolutePath}.") } } } @@ -183,6 +184,8 @@ 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}") } } diff --git a/core/src/main/scala/kafka/log/TimeIndex.scala b/core/src/main/scala/kafka/log/TimeIndex.scala index e505f36..e75e3e5 100644 --- a/core/src/main/scala/kafka/log/TimeIndex.scala +++ b/core/src/main/scala/kafka/log/TimeIndex.scala @@ -57,6 +57,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()}") + // 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 @@ -86,7 +89,8 @@ class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: def entry(n: Int): TimestampOffset = { maybeLock(lock) { if(n >= _entries) - throw new IllegalArgumentException("Attempt to fetch the %dth entry from a time index of size %d.".format(n, _entries)) + throw new IllegalArgumentException(s"Attempt to fetch the ${n}th entry from time index ${file.getAbsolutePath} " + + s"which has size ${_entries}.") val idx = mmap.duplicate TimestampOffset(timestamp(idx, n), relativeOffset(idx, n)) } @@ -117,16 +121,16 @@ class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: // 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("Attempt to append an offset (%d) to slot %d no larger than the last offset appended (%d) to %s." - .format(offset, _entries, lastEntry.offset, file.getAbsolutePath)) + 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("Attempt to append a timestamp (%d) to slot %d no larger than the last timestamp appended (%d) to %s." - .format(timestamp, _entries, lastEntry.timestamp, file.getAbsolutePath)) + 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) { - debug("Adding index entry %d => %d to %s.".format(timestamp, offset, file.getName)) + trace(s"Adding index entry $timestamp => $offset to ${file.getAbsolutePath}.") mmap.putLong(timestamp) mmap.putInt((offset - baseOffset).toInt) _entries += 1 @@ -202,6 +206,7 @@ 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}") } } diff --git a/core/src/main/scala/kafka/log/TransactionIndex.scala b/core/src/main/scala/kafka/log/TransactionIndex.scala index 349f0ce..ec59225 100644 --- a/core/src/main/scala/kafka/log/TransactionIndex.scala +++ b/core/src/main/scala/kafka/log/TransactionIndex.scala @@ -53,7 +53,8 @@ class TransactionIndex(val startOffset: Long, @volatile var file: File) extends def append(abortedTxn: AbortedTxn): Unit = { lastOffset.foreach { offset => if (offset >= abortedTxn.lastOffset) - throw new IllegalArgumentException("The last offset of appended transactions must increase sequentially") + throw new IllegalArgumentException(s"The last offset of appended transactions must increase sequentially, but " + + s"${abortedTxn.lastOffset} is not greater than current last offset $offset of index ${file.getAbsolutePath}") } lastOffset = Some(abortedTxn.lastOffset) Utils.writeFully(channel, abortedTxn.buffer.duplicate()) @@ -138,8 +139,8 @@ class TransactionIndex(val startOffset: Long, @volatile var file: File) extends val abortedTxn = new AbortedTxn(buffer) if (abortedTxn.version > AbortedTxn.CurrentVersion) - throw new KafkaException(s"Unexpected aborted transaction version ${abortedTxn.version}, " + - s"current version is ${AbortedTxn.CurrentVersion}") + throw new KafkaException(s"Unexpected aborted transaction version ${abortedTxn.version} " + + s"in transaction index ${file.getAbsolutePath}, current version is ${AbortedTxn.CurrentVersion}") val nextEntry = (abortedTxn, position) position += AbortedTxn.TotalSize nextEntry @@ -147,7 +148,7 @@ class TransactionIndex(val startOffset: Long, @volatile var file: File) extends case e: IOException => // We received an unexpected error reading from the index file. We propagate this as an // UNKNOWN error to the consumer, which will cause it to retry the fetch. - throw new KafkaException(s"Failed to read from the transaction index $file", e) + throw new KafkaException(s"Failed to read from the transaction index ${file.getAbsolutePath}", e) } } } @@ -187,8 +188,8 @@ class TransactionIndex(val startOffset: Long, @volatile var file: File) extends val buffer = ByteBuffer.allocate(AbortedTxn.TotalSize) for ((abortedTxn, _) <- iterator(() => buffer)) { if (abortedTxn.lastOffset < startOffset) - throw new CorruptIndexException(s"Last offset of aborted transaction $abortedTxn is less than start offset " + - s"$startOffset") + throw new CorruptIndexException(s"Last offset of aborted transaction $abortedTxn in index " + + s"${file.getAbsolutePath} is less than start offset $startOffset") } }