[GitHub] [kafka] dhruvilshah3 commented on a change in pull request #10763: KAFKA-12520: Ensure log loading does not truncate producer state unless required

[GitBox]

dhruvilshah3 commented on a change in pull request #10763:
URL: https://github.com/apache/kafka/pull/10763#discussion_r652157061



##########
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)

Review comment:
       You could perhaps define a method like this:
   
   ```
         def maybeCompleteInterruptedSwap(fn: (File, Long, String) => File): 
Boolean = {
           val swapIndexFile = fn(swapFile.getParentFile, baseOffset, 
Log.SwapFileSuffix)
           if (!swapIndexFile.exists()) {
             val cleanedIndexFile = fn(swapFile.getParentFile, baseOffset, 
Log.CleanedFileSuffix)
             if (cleanedIndexFile.exists()) {
               cleanedIndexFile.renameTo(swapIndexFile)
               true
             } else {
               false
             }
           }
         }
   ```
   
   and then invoke it as
   
   ```
         var recoverable = true
         recoverable = maybeCompleteInterruptedSwap(Log.offsetIndexFile)
         if (recoverable)
           recoverable = maybeCompleteInterruptedSwap(Log.timeIndexFile)
         if (recoverable)
           recoverable = maybeCompleteInterruptedSwap(Log.transactionIndexFile)
   ```




-- 
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