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

[GitBox]

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



##########
File path: core/src/main/scala/kafka/log/LogLoader.scala
##########
@@ -192,20 +230,10 @@ object LogLoader extends Logging {
         debug(s"${params.logIdentifier}Deleting stray temporary file 
${file.getAbsolutePath}")
         Files.deleteIfExists(file.toPath)
       } else if (filename.endsWith(CleanedFileSuffix)) {
-        minCleanedFileOffset = Math.min(offsetFromFileName(filename), 
minCleanedFileOffset)
-        cleanFiles += file
+        minCleanedFileOffset = Math.min(offsetFromFile(file), 
minCleanedFileOffset)
+        cleanedFiles += file
       } else if (filename.endsWith(SwapFileSuffix)) {
-        // we crashed in the middle of a swap operation, to recover:
-        // if a log, delete the index files, complete the swap operation later
-        // if an index just delete the index files, they will be rebuilt
-        val baseFile = new File(CoreUtils.replaceSuffix(file.getPath, 
SwapFileSuffix, ""))
-        info(s"${params.logIdentifier}Found file ${file.getAbsolutePath} from 
interrupted swap operation.")
-        if (Log.isIndexFile(baseFile)) {
-          deleteIndicesIfExist(baseFile)
-        } else if (Log.isLogFile(baseFile)) {
-          deleteIndicesIfExist(baseFile)
-          swapFiles += file
-        }
+        swapFiles += file

Review comment:
       Thanks for the explanation. Make sense.




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

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org