Skip to content

KAFKA-14475: Move TimeIndex/LazyIndex to storage module #13010

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 3 commits into from
Dec 21, 2022
Merged

KAFKA-14475: Move TimeIndex/LazyIndex to storage module #13010

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a776107
Move TimeIndex to storage module
ijuma Dec 12, 2022
9294e30
Move LazyIndex to storage module
ijuma Dec 19, 2022
4ae7ae9
Remove unused TimeIndex constructor
ijuma Dec 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 0 additions & 166 deletions core/src/main/scala/kafka/log/LazyIndex.scala
View file
Open in desktop

This file was deleted.

13 changes: 6 additions & 7 deletions core/src/main/scala/kafka/log/LogSegment.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ import org.apache.kafka.common.errors.CorruptRecordException
import org.apache.kafka.common.record.FileRecords.{LogOffsetPosition, TimestampAndOffset}
import org.apache.kafka.common.record._
import org.apache.kafka.common.utils.{BufferSupplier, Time}
import org.apache.kafka.server.log.internals.{AbortedTxn, CompletedTxn, OffsetIndex, OffsetPosition, TimestampOffset, TransactionIndex, TxnIndexSearchResult}
import org.apache.kafka.server.log.internals.{AbortedTxn, CompletedTxn, LazyIndex, OffsetIndex, OffsetPosition, TimeIndex, TimestampOffset, TransactionIndex, TxnIndexSearchResult}

import java.util.Optional
import scala.jdk.CollectionConverters._
Expand Down Expand Up @@ -381,7 +381,7 @@ class LogSegment private[log] (val log: FileRecords,
log.truncateTo(validBytes)
offsetIndex.trimToValidSize()
// A normally closed segment always appends the biggest timestamp ever seen into log segment, we do this as well.
timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true)
timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, true)
timeIndex.trimToValidSize()
truncated
}
Expand Down Expand Up @@ -511,7 +511,7 @@ class LogSegment private[log] (val log: FileRecords,
* The time index entry appended will be used to decide when to delete the segment.
*/
def onBecomeInactiveSegment(): Unit = {
timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true)
timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, true)
offsetIndex.trimToValidSize()
timeIndex.trimToValidSize()
log.trim()
Expand Down Expand Up @@ -593,8 +593,7 @@ class LogSegment private[log] (val log: FileRecords,
*/
def close(): Unit = {
if (_maxTimestampAndOffsetSoFar != TimestampOffset.UNKNOWN)
CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar,
skipFullCheck = true), this)
CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, true), this)
CoreUtils.swallow(lazyOffsetIndex.close(), this)
CoreUtils.swallow(lazyTimeIndex.close(), this)
CoreUtils.swallow(log.close(), this)
Expand Down Expand Up @@ -673,8 +672,8 @@ object LogSegment {
val maxIndexSize = config.maxIndexSize
new LogSegment(
FileRecords.open(UnifiedLog.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate),
LazyIndex.forOffset(UnifiedLog.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
LazyIndex.forTime(UnifiedLog.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
LazyIndex.forOffset(UnifiedLog.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset, maxIndexSize, true),
LazyIndex.forTime(UnifiedLog.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset, maxIndexSize, true),
new TransactionIndex(baseOffset, UnifiedLog.transactionIndexFile(dir, baseOffset, fileSuffix)),
baseOffset,
indexIntervalBytes = config.indexInterval,
Expand Down
Loading