Skip to content

KAFKA-10101: Fix edge cases in Log.recoverLog and LogManager.loadLogs #8812

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 4 commits into from
Feb 26, 2021
Merged

KAFKA-10101: Fix edge cases in Log.recoverLog and LogManager.loadLogs #8812

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
91de172
KAFKA-10101: Fix edge cases in Log.recoverLog and LogManager.loadLogs
ijuma Feb 25, 2021
7b6abd6
Address feedback and resolve FIXME
ijuma Feb 25, 2021
796dd8b
Tweak logic
ijuma Feb 25, 2021
46d5cef
Fix failing test
ijuma Feb 25, 2021
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
46 changes: 33 additions & 13 deletions core/src/main/scala/kafka/log/Log.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -849,9 +849,25 @@ class Log(@volatile private var _dir: File,
* @throws LogSegmentOffsetOverflowException if we encountered a legacy segment with offset overflow
*/
private[log] def recoverLog(): Long = {
/** return the log end offset if valid */
def deleteSegmentsIfLogStartGreaterThanLogEnd(): Option[Long] = {
if (logSegments.nonEmpty) {
val logEndOffset = activeSegment.readNextOffset
if (logEndOffset >= logStartOffset)
Some(logEndOffset)
else {
warn(s"Deleting all segments because logEndOffset ($logEndOffset) is smaller than logStartOffset ($logStartOffset). " +
"This could happen if segment files were deleted from the file system.")
removeAndDeleteSegments(logSegments, asyncDelete = true, LogRecovery)
leaderEpochCache.foreach(_.clearAndFlush())
producerStateManager.truncateFullyAndStartAt(logStartOffset)
None
}
} else None
}

// if we have the clean shutdown marker, skip recovery
if (!hadCleanShutdown) {
// okay we need to actually recover this log
val unflushed = logSegments(this.recoveryPoint, Long.MaxValue).iterator
var truncated = false

Expand Down Expand Up @@ -879,16 +895,7 @@ class Log(@volatile private var _dir: File,
}
}

if (logSegments.nonEmpty) {
val logEndOffset = activeSegment.readNextOffset
if (logEndOffset < logStartOffset) {
warn(s"Deleting all segments because logEndOffset ($logEndOffset) is smaller than logStartOffset ($logStartOffset). " +
"This could happen if segment files were deleted from the file system.")
removeAndDeleteSegments(logSegments,
asyncDelete = true,
reason = LogRecovery)
}
}
val logEndOffsetOption = deleteSegmentsIfLogStartGreaterThanLogEnd()

if (logSegments.isEmpty) {
// no existing segments, create a new mutable segment beginning at logStartOffset
Expand All @@ -900,8 +907,21 @@ class Log(@volatile private var _dir: File,
preallocate = config.preallocate))
}

recoveryPoint = activeSegment.readNextOffset

@hachikuji hachikuji Jun 5, 2020

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.

Can you help me understand what was wrong with this?

@ijuma ijuma Jun 5, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Jun explained in the JIRA, The concern is that if there is a hard failure during recovery, you could end up with a situation where we persisted this, but we did not flush some of the segments. Does that make sense?

@hachikuji hachikuji Jun 5, 2020

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.

Got it. So we may still be able to reopen an unflushed segment. That makes sense.

@purplefox purplefox Feb 26, 2021 •
edited
Loading

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.

Is it possible that a consumer could see "phantom" messages after recovery, even with this change?

  1. Kafka Process dies with log data in page cache but not fsync'd
  2. Recovery process sees the un-fsync'd log data but it looks ok so recovery succeeds, nothing to do.
  3. Consumer fetches this data
  4. OS hard dies, losing page cache
  5. Broker is restarted and consumer tries to repeat fetch from same offset but data has gone.

It seems to me once recovery has run we should be sure that all log segments are persistently stored. I'm not sure if we're currently providing that guarantee. It would be pretty simple just to fsync each segment we recover.

@ijuma ijuma Feb 26, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

@purplefox To clarify, the case you are talking about is:

  1. A catastrophic scenario where a partition is offline (i.e. all replicas are down)
  2. The OS was not shutdown
  3. The OS did not flush the data to disk (this typically happens irrespective of our flushes due to OS configurations)
  4. The replica that was the last member of the ISR comes back up, registers, to ZK and the Controller makes it the
    leader (since it was the last member of the ISR, if it was a different replica, it won't be given leadership without
    unclean leader election)
  5. The hw is beyond the flushed segments
  6. Consumer fetches the data beyond the flushed segments
  7. OS hard dies

This is an interesting edge case, it seems incredibly unlikely, but possible if the hw can be beyond the flushed segments. @junrao & @hachikuji are we missing any detail that protects us against this?

The following has a similar impact:

  1. Leader accepts a write
  2. Write is replicated, hw is incremented, but data is not flushed
  3. All replicas die, but the ISR is not shrunk yet
  4. Leader receives a write, accepts it, replication doesn't happen since replicas are gone
  5. ISR is shrunk, hw is incremented
  6. Producer won't receive a successful ack given min.isr=2, but the consumer reads data that is only in the leader
  7. Leader crashes and the unflushed data is gone (or hard disk dies and all the data in the leader is gone)

Flushing the segments during recovery helps on some scenarios, but not the one I just mentioned (assuming I am not missing anything). @hachikuji had a "strict min isr" proposal where the ISR is never allowed to shrink below min.isr. I haven't thought about all the details, but perhaps that covers both issues. Thoughts @hachikuji?

@junrao junrao Feb 26, 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.

For the case that Tim mentioned, if we defer advancing the recovery point, at step 5, the broker will be forced to do log recovery for all unflushed data. If the data is corrupted on disk, it will be detected during recovery.

For the other case that Ismael mentioned, it is true that data can be lost in that case, but then this is the case where all replicas have failed.

@hachikuji hachikuji Feb 26, 2021 •
edited
Loading

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 think it is a gap that there is no minimum replication factor before a write can get exposed. Any writes that end up seeing the NOT_ENOUGH_REPLICAS_AFTER_APPEND error code are more vulnerable. These are unacknowledged writes, and the producer is expected to retry, but the consumer can still read them once the ISR shrinks and we would still view it as "data loss" if the broker failed before they could be flushed to disk. With the "strict min isr" proposal, the leader is not allowed to shrink the ISR lower than some replication factor, which helps to plug this hole.

Going back to @purplefox's suggestion, it does seem like a good idea to flush segments beyond the recovery point during recovery. It kind of serves to constrain the initial state of the system which makes it easier to reason about (e.g. you only need to worry about the loss of unflushed data from the last restart). Some of the flush weaknesses probably still exist though regardless of this change.

@ijuma ijuma Feb 26, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

We discussed this offline and we decided to stick with the fix in this PR for now and to file a separate JIRA to consider flushing unflushed segments during recovery. That would provide stronger guarantees after a restart.

@ijuma ijuma Feb 27, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Filed https://issues.apache.org/jira/browse/KAFKA-12386.

recoveryPoint
// Update the recovery point if there was a clean shutdown and did not perform any changes to
// the segment. Otherwise, we just ensure that the recovery point is not ahead of the log end
// offset. To ensure correctness and to make it easier to reason about, it's best to only advance
// the recovery point in flush(Long). If we advanced the recovery point here, we could skip recovery for
// unflushed segments if the broker crashed after we checkpoint the recovery point and before we flush the
// segment.
(hadCleanShutdown, logEndOffsetOption) match {
case (true, Some(logEndOffset)) =>
recoveryPoint = logEndOffset
logEndOffset
case _ =>
val logEndOffset = logEndOffsetOption.getOrElse(activeSegment.readNextOffset)
recoveryPoint = Math.min(recoveryPoint, logEndOffset)
logEndOffset
}
}

// Rebuild producer state until lastOffset. This method may be called from the recovery code path, and thus must be
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/log/LogManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -318,7 +318,7 @@ class LogManager(logDirs: Seq[File],
info(s"Skipping recovery for all logs in $logDirAbsolutePath since clean shutdown file was found")
// Cache the clean shutdown status and use that for rest of log loading workflow. Delete the CleanShutdownFile
// so that if broker crashes while loading the log, it is considered hard shutdown during the next boot up. KAFKA-10471
cleanShutdownFile.delete()
Files.deleteIfExists(cleanShutdownFile.toPath)
hadCleanShutdown = true
} else {
// log recovery itself is being performed by `Log` class during initialization
Expand Down
6 changes: 5 additions & 1 deletion core/src/test/scala/unit/kafka/log/LogTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2518,7 +2518,11 @@ class LogTest {
log.close()

// test recovery case
log = createLog(logDir, logConfig, lastShutdownClean = false)
val recoveryPoint = 10
log = createLog(logDir, logConfig, recoveryPoint = recoveryPoint, lastShutdownClean = false)
// the recovery point should not be updated after unclean shutdown until the log is flushed
verifyRecoveredLog(log, recoveryPoint)
log.flush()
verifyRecoveredLog(log, lastOffset)
log.close()
}
Expand Down