ccding commented on a change in pull request #10763:
URL: https://github.com/apache/kafka/pull/10763#discussion_r652151007
##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -368,12 +368,50 @@ object LogLoader extends Logging {
for (swapFile <- swapFiles) {
val logFile = new File(CoreUtils.replaceSuffix(swapFile.getPath,
Log.SwapFileSuffix, ""))
val baseOffset = Log.offsetFromFile(logFile)
+ // Check whether swap index files exist: if not, the cleaned files must
exist due to the
+ // existence of swap log file. Therefore, we rename the cleaned files to
swap files and continue.
+ var recoverable = true
+ val swapOffsetIndexFile = Log.offsetIndexFile(swapFile.getParentFile,
baseOffset, Log.SwapFileSuffix)
+ if (!swapOffsetIndexFile.exists()) {
+ val cleanedOffsetIndexFile =
Log.offsetIndexFile(swapFile.getParentFile, baseOffset, Log.CleanedFileSuffix)
+ if (cleanedOffsetIndexFile.exists())
+ cleanedOffsetIndexFile.renameTo(swapOffsetIndexFile)
+ else
+ recoverable = false
+ }
+ val swapTimeIndexFile = Log.timeIndexFile(swapFile.getParentFile,
baseOffset, Log.SwapFileSuffix)
+ if (!swapTimeIndexFile.exists()) {
+ val cleanedTimeIndexFile = Log.timeIndexFile(swapFile.getParentFile,
baseOffset, Log.CleanedFileSuffix)
+ if (cleanedTimeIndexFile.exists())
+ cleanedTimeIndexFile.renameTo(swapTimeIndexFile)
+ else
+ recoverable = false
+ }
+ val swapTxnIndexFile = Log.transactionIndexFile(swapFile.getParentFile,
baseOffset, Log.SwapFileSuffix)
+ if (!swapTxnIndexFile.exists()) {
+ val cleanedTxnIndexFile =
Log.transactionIndexFile(swapFile.getParentFile, baseOffset,
Log.CleanedFileSuffix)
+ if (cleanedTxnIndexFile.exists())
+ cleanedTxnIndexFile.renameTo(swapTxnIndexFile)
+ else
+ recoverable = false
+ }
val swapSegment = LogSegment.open(swapFile.getParentFile,
baseOffset = baseOffset,
params.config,
time = params.time,
fileSuffix = Log.SwapFileSuffix)
- info(s"${params.logIdentifier}Found log file ${swapFile.getPath} from
interrupted swap operation, repairing.")
+ if (recoverable) {
Review comment:
The whole logic is that, if the segment.swap file exists, then all index
files should exist as .cleaned or .swap. We find them and rename them to .swap
[before this block of code]. Then do a sanity check and rename all the .swap
files to non-suffix log files [within this block of code].
This could fix the issue caused by the compaction as we discussed before.
For all other cases, I think it is in an inconsistent state and we will have
to do the original recovery.
Does this make sense to you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
