Skip to content

KAFKA-13603: Allow the empty active segment to have missing offset index during recovery #11345

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
junrao merged 33 commits into from
Jan 27, 2022
Merged

KAFKA-13603: Allow the empty active segment to have missing offset index during recovery #11345

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
33 commits
Select commit Hold shift + click to select a range
ecb8692
Allow empty last segment to have missing offset index during recovery
ccding Sep 20, 2021
9904830
Revert "Allow empty last segment to have missing offset index during …
ccding Oct 1, 2021
ac6e9cd
Merge branch 'trunk' into last
ccding Oct 1, 2021
a2c51d9
flush empty active segments
ccding Oct 1, 2021
82913f0
add comment
ccding Oct 26, 2021
b915fe2
unit test
ccding Oct 27, 2021
4df64d8
address comments
ccding Nov 5, 2021
0f2f0a8
Merge branch 'trunk' into last
ccding Nov 8, 2021
c50bff2
flush at the right place
ccding Nov 17, 2021
603627f
Merge branch 'trunk' into last
ccding Nov 17, 2021
56d3f06
trigger test
ccding Nov 17, 2021
425a7a7
address comments
ccding Nov 19, 2021
f345831
Merge branch 'trunk' into last
ccding Nov 22, 2021
069ac5b
only do the inclusive flush in KRaft.close()
ccding Nov 22, 2021
049c856
Merge branch 'trunk' into last
ccding Nov 30, 2021
56d64cb
rename a bunch of variable/function names to address comments
ccding Nov 30, 2021
7bb37b8
fix typo and purpose
ccding Dec 1, 2021
40506eb
fix
ccding Dec 1, 2021
6f70af3
do not print error if the missing index file is after the recovery point
ccding Dec 7, 2021
23314ea
add log loader test
ccding Dec 20, 2021
d775f14
fix time stamp check
ccding Dec 20, 2021
4d88092
fix comment
ccding Dec 20, 2021
099cb42
fix log loader test
ccding Dec 21, 2021
229a537
Merge branch 'trunk' into last
ccding Dec 22, 2021
14239c0
trigger test
ccding Dec 22, 2021
8b7e0c9
Merge branch 'trunk' into last
ccding Dec 28, 2021
0c8e48b
Merge branch 'trunk' into last
ccding Jan 5, 2022
25821db
address comments
ccding Jan 10, 2022
47ba1d5
update comments and polish log messages
ccding Jan 11, 2022
885db15
improve log message
ccding Jan 24, 2022
5bb1a47
Merge branch 'trunk' into last
ccding Jan 27, 2022
e00906a
fix compile and improve log message
ccding Jan 27, 2022
bd21bb2
fix debug output
ccding Jan 27, 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
6 changes: 4 additions & 2 deletions core/src/main/scala/kafka/log/LogLoader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,7 +310,9 @@ object LogLoader extends Logging {
private def loadSegmentFiles(params: LoadLogParams): Unit = {
// load segments in ascending order because transactional data from one segment may depend on the
// segments that come before it
for (file <- params.dir.listFiles.sortBy(_.getName) if file.isFile) {
val files = params.dir.listFiles.filter(_.isFile).sortBy(_.getName)
val lastLogFileOpt = files.filter(isLogFile).lastOption
for (file <- files) {
if (isIndexFile(file)) {
// if it is an index file, make sure it has a corresponding .log file
val offset = offsetFromFile(file)
Expand All @@ -330,7 +332,7 @@ object LogLoader extends Logging {
time = params.time,
fileAlreadyExists = true)

try segment.sanityCheck(timeIndexFileNewlyCreated)
try segment.sanityCheck(timeIndexFileNewlyCreated, isActiveSegment = file == lastLogFileOpt.get)
catch {
case _: NoSuchFileException =>
error(s"${params.logIdentifier}Could not find offset index file corresponding to log file" +
Expand Down
5 changes: 3 additions & 2 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 @@ -77,8 +77,9 @@ class LogSegment private[log] (val log: FileRecords,
timeIndex.resize(size)
}

def sanityCheck(timeIndexFileNewlyCreated: Boolean): Unit = {
if (lazyOffsetIndex.file.exists) {
def sanityCheck(timeIndexFileNewlyCreated: Boolean, isActiveSegment: Boolean): Unit = {
// We allow for absence of offset index file only for an empty active segment.
if ((isActiveSegment && size == 0) || lazyOffsetIndex.file.exists) {

@junrao junrao Sep 29, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am wondering why the active segment will be missing the offset index file during a clean shutdown. When we load the segments during broker restart, we call resizeIndexes() on the last segment. This should trigger the creation of the offset index file, which will be flushed on broker shutdown.

@ccding ccding Sep 29, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When we load the segments during broker restart, we call resizeIndexes() on the last segment. This should trigger the creation of the offset index file, which will be flushed on broker shutdown.

The sanityCheck is called before resizeIndexes.

It appears you are talking about we start the broker then immediately shut it down. In between the active segment may have been changed, and if the new one is empty, no index file is created.

@junrao junrao Sep 29, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am still trying to understand if the missing index is the result of a clean shutdown or a hard shutdown. When will roll a segment, the index on the new active segment is created lazily. However, during a clean shutdown, we force flush the active segment, which should trigger the creation of an empty index file because the following method is used in segment flush.

def offsetIndex: OffsetIndex = lazyOffsetIndex.get
On a hard shutdown, it's possible for the offset index to be missing. However, in that case, the offset index can be missing even when the log is not empty. So, I am wondering how common of an issue that we are fixing.

@kowshik kowshik Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@junrao: When the UnifiedLog is flushed during clean shutdown, we flush the LocaLog until the logEndOffset. Here an empty active segment is not included in the list of candidate segments to be flushed. The reason is that during LocalLog.flush(), the LogSegments.values(recoveryPoint, logEndOffset) call here does not select the empty active segment (doc), because, the logEndOffset would match the base offset of the empty active segment and thus get ommitted. So, prior to clean shutdown if the empty active segment's offset index was never created before, then, the offset index will not be created during clean shutdown because the empty active segment is never flushed.

The above is shown in the following passing unit test:

@Test
def testFlushEmptyActiveSegmentDoesNotCreateOffsetIndex(): Unit = {
    // Create an empty log.
    val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024)
    val log = createLog(logDir, logConfig)
    val oneRecord = TestUtils.records(List(
      new SimpleRecord(mockTime.milliseconds, "a".getBytes, "value".getBytes)
    ))

    // Append a record and flush. Verify that there exists only 1 segment.
    log.appendAsLeader(oneRecord, leaderEpoch = 0)
    assertEquals(1, log.logEndOffset)
    log.flush()
    assertEquals(1, log.logSegments.size)
    assertTrue(UnifiedLog.logFile(logDir, 0).exists())
    assertTrue(UnifiedLog.offsetIndexFile(logDir, 0).exists())
    assertFalse(UnifiedLog.logFile(logDir, 1).exists())
    assertFalse(UnifiedLog.offsetIndexFile(logDir, 1).exists())

    // Roll the log and verify that the new active segment's offset index is missing.
    log.roll()
    assertEquals(2, log.logSegments.size)
    assertTrue(UnifiedLog.logFile(logDir, 0).exists())
    assertTrue(UnifiedLog.offsetIndexFile(logDir, 0).exists())
    assertTrue(UnifiedLog.logFile(logDir, 1).exists())
    assertFalse(UnifiedLog.offsetIndexFile(logDir, 1).exists())

    // Flush the log and once again verify that the active segment's offset index is still missing.
    log.flush()
    assertTrue(UnifiedLog.logFile(logDir, 0).exists())
    assertTrue(UnifiedLog.offsetIndexFile(logDir, 0).exists())
    assertTrue(UnifiedLog.logFile(logDir, 1).exists())
    assertFalse(UnifiedLog.offsetIndexFile(logDir, 1).exists())

    // Close the log and verify that the active segment's offset index is still missing.
    log.close()
    assertTrue(UnifiedLog.logFile(logDir, 0).exists())
    assertTrue(UnifiedLog.offsetIndexFile(logDir, 0).exists())
    assertTrue(UnifiedLog.logFile(logDir, 1).exists())
    assertFalse(UnifiedLog.offsetIndexFile(logDir, 1).exists())
}

This PR mainly fixes a logging issue in the code. For example, one situation where the issue happens more frequently is the following: Imagine there exists a topic with very low ingress traffic in some/all partitions. Imagine that for this topic the retention setting causes all existing segments to expire and get removed. In such a case, we roll the log to create an active segment. This ensures there is at least one segment remaining in the LocalLog when the retention loop completes. However we don't create the offset index for the active segment until the first append operation. Now before the first append, if the Kafka cluster is rolled then we will see this false negative corruption error message during recovery.

This PR fixes the logging problem by ignoring the absence of offset index for an empty active segment during recovery.

@junrao junrao Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kowshik : Thanks for the great explanation. It makes sense to me now.

If we don't flush the only empty log segment during clean shutdown, we could lose the log segment file as well, which causes the replica to lose track of logEndOffset. I am wondering if we should force flush the empty log segment during clean shutdown too.

@ccding ccding Sep 30, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will def flush(): Unit = flush(logEndOffset + 1) trigger flushing empty active segments every time we roll a segment, not only during shutdown?

@kowshik kowshik Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ccding When we roll a segment, we explicitly ensure to flush only the old segment. See this LOC.

@junrao junrao Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kowshik : #8346 tries to avoid opening the index during close() when the index is not opened yet. This applies to existing segments on broker restart. For active segment, we typically need to open the index anyway. So, we probably don't need to optimize the rare case when it's empty. Plus, the danger of losing logEndOffset is a bigger concern than avoiding the cost of opening one index file.

@kowshik kowshik Oct 1, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@junrao Sounds good to me. We can flush the active segment during clean shutdown. That's a very elegant way to handle this problem.

@ccding Would you like to update the PR with the approach proposed by @junrao?

@ccding ccding Oct 1, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated. Please let me know if I misunderstood anything.

// Resize the time index file to 0 if it is newly created.
if (timeIndexFileNewlyCreated)
timeIndex.resize(0)
Expand Down
47 changes: 46 additions & 1 deletion core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
package kafka.log

import java.io.File

import kafka.server.checkpoints.LeaderEpochCheckpoint
import kafka.server.epoch.EpochEntry
import kafka.server.epoch.LeaderEpochFileCache
Expand All @@ -29,6 +28,7 @@ import org.apache.kafka.common.utils.{MockTime, Time, Utils}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}

import java.nio.file.NoSuchFileException
import scala.jdk.CollectionConverters._
import scala.collection._
import scala.collection.mutable.ArrayBuffer
Expand Down Expand Up @@ -586,4 +586,49 @@ class LogSegmentTest {
Utils.delete(tempDir)
}

@Test
def testSanityCheckThrowsDuringMissingOffsetIndex(): Unit = {
// Missing offset index for non-active segment should throw
val seg = createSegment(0)
assertFalse(seg.lazyOffsetIndex.file.exists())
assertTrue(seg.size == 0)
for (timeIndexFileNewlyCreated <- Array(true, false)) {
assertThrows(classOf[NoSuchFileException], () => seg.sanityCheck(timeIndexFileNewlyCreated = timeIndexFileNewlyCreated, isActiveSegment = false))
}

// Missing offset index for non-empty active segment should throw
seg.append(0, RecordBatch.NO_TIMESTAMP, -1L, LogTestUtils.records(0, "hello"))
assertTrue(seg.size > 0)
val offsetIndex = seg.lazyOffsetIndex.get
assertTrue(offsetIndex.file.exists())
offsetIndex.deleteIfExists()
for (timeIndexFileNewlyCreated <- Array(true, false)) {
assertThrows(classOf[NoSuchFileException], () => seg.sanityCheck(timeIndexFileNewlyCreated = timeIndexFileNewlyCreated, isActiveSegment = true))
}
}

@Test
def testSanityCheckResizesTimeIndex(): Unit = {
val seg = createSegment(0)
assertFalse(LocalLog.timeIndexFile(logDir, 0).exists())
// Create the lazy offset index to bypass the file existence check in LogSegment.sanityCheck()
seg.lazyOffsetIndex.get

seg.sanityCheck(timeIndexFileNewlyCreated = true, isActiveSegment = false)

assertTrue(LocalLog.timeIndexFile(logDir, 0).exists())
assertTrue(seg.timeIndex.sizeInBytes == 0)
}

@Test
def testSanityCheckTransactionIndexCheck(): Unit = {
val seg = createSegment(100)
seg.append(100, RecordBatch.NO_TIMESTAMP, -1L, LogTestUtils.records(100, "hello"))
assertTrue(seg.lazyTimeIndex.file.exists())
assertTrue(seg.lazyOffsetIndex.file.exists())
seg.sanityCheck(timeIndexFileNewlyCreated = false, isActiveSegment = false)
// Intentionally corrupt the transaction index causing LogSegment.sanityCheck() to throw
seg.txnIndex.append(new AbortedTxn(producerId = 0L, firstOffset = 0, lastOffset = 10, lastStableOffset = 11))
assertThrows(classOf[CorruptIndexException], () => seg.sanityCheck(timeIndexFileNewlyCreated = false, isActiveSegment = false))
}
}
8 changes: 7 additions & 1 deletion core/src/test/scala/unit/kafka/log/LogTestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import kafka.server.checkpoints.LeaderEpochCheckpointFile
import kafka.server.{BrokerTopicStats, FetchDataInfo, FetchIsolation, FetchLogEnd, LogDirFailureChannel}
import kafka.utils.{Scheduler, TestUtils}
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.record.{CompressionType, ControlRecordType, EndTransactionMarker, FileRecords, MemoryRecords, RecordBatch, SimpleRecord}
import org.apache.kafka.common.record.{CompressionType, ControlRecordType, EndTransactionMarker, FileRecords, MemoryRecords, RecordBatch, SimpleRecord, TimestampType}
import org.apache.kafka.common.utils.{Time, Utils}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse}

Expand All @@ -46,6 +46,12 @@ object LogTestUtils {
new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time)
}

/* create a ByteBufferMessageSet for the given messages starting from the given offset */
def records(offset: Long, records: String*): MemoryRecords = {
MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V1, offset, CompressionType.NONE, TimestampType.CREATE_TIME,
records.map { s => new SimpleRecord(offset * 10, s.getBytes) }: _*)
}

def createLogConfig(segmentMs: Long = Defaults.SegmentMs,
segmentBytes: Int = Defaults.SegmentSize,
retentionMs: Long = Defaults.RetentionMs,
Expand Down