More feedback.

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -452,7 +452,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
             override def run(): Unit = mergeApplicationListing(entry, newLastScanTime)
           })
         } catch {
-          // let the iteration over logInfos break, since an exception on
+          // let the iteration over the updated entries break, since an exception on
           // replayExecutor.submit (..) indicates the ExecutorService is unable
           // to take any more submissions at this time
           case e: Exception =>
@@ -636,6 +636,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
         (Some(app.info.id), app.attempts.head.info.attemptId)
 
       case _ =>
+        // If the app hasn't written down its app ID to the logs, still record the entry in the
+        // listing db, with an empty ID. This will make the log eligible for deletion if the app
+        // does not make progress after the configured max log age.
         (None, None)
     }
     listing.write(LogInfo(logPath.toString(), scanTime, appId, attemptId, fileStatus.getLen()))
@@ -904,6 +907,11 @@ private[history] case class FsHistoryProviderMetadata(
     uiVersion: Long,
     logDir: String)
 
+/**
+ * Tracking info for event logs detected in the configured log directory. Tracks both valid and
+ * invalid logs (e.g. unparseable logs, recorded as logs with no app ID) so that the cleaner
+ * can know what log files are safe to delete.
+ */
 private[history] case class LogInfo(
     @KVIndexParam logPath: String,
     @KVIndexParam("lastProcessed") lastProcessed: Long,

core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala

@@ -716,42 +716,62 @@ class FsHistoryProviderSuite extends SparkFunSuite with BeforeAndAfter with Matc
   }
 
   test("SPARK-21571: clean up removes invalid history files") {
+    // TODO: "maxTime" becoming negative in cleanLogs() causes this test to fail, so avoid that
+    // until we figure out what's causing the problem.
     val clock = new ManualClock(TimeUnit.DAYS.toMillis(120))
-    val conf = createTestConf().set("spark.history.fs.cleaner.maxAge", s"2d")
+    val conf = createTestConf().set(MAX_LOG_AGE_S.key, s"2d")
     val provider = new FsHistoryProvider(conf, clock) {
       override def getNewLastScanTime(): Long = clock.getTimeMillis()
     }
 
     // Create 0-byte size inprogress and complete files
-    val logfile1 = newLogFile("emptyInprogressLogFile", None, inProgress = true)
-    logfile1.createNewFile()
-    logfile1.setLastModified(clock.getTimeMillis())
+    var logCount = 0
+    var validLogCount = 0
 
-    val logfile2 = newLogFile("emptyFinishedLogFile", None, inProgress = false)
-    logfile2.createNewFile()
-    logfile2.setLastModified(clock.getTimeMillis())
+    val emptyInProgress = newLogFile("emptyInprogressLogFile", None, inProgress = true)
+    emptyInProgress.createNewFile()
+    emptyInProgress.setLastModified(clock.getTimeMillis())
+    logCount += 1
+
+    val slowApp = newLogFile("slowApp", None, inProgress = true)
+    slowApp.createNewFile()
+    slowApp.setLastModified(clock.getTimeMillis())
+    logCount += 1
+
+    val emptyFinished = newLogFile("emptyFinishedLogFile", None, inProgress = false)
+    emptyFinished.createNewFile()
+    emptyFinished.setLastModified(clock.getTimeMillis())
+    logCount += 1
 
     // Create an incomplete log file, has an end record but no start record.
-    val logfile3 = newLogFile("nonEmptyCorruptLogFile", None, inProgress = false)
-    writeFile(logfile3, true, None, SparkListenerApplicationEnd(0))
-    logfile3.setLastModified(clock.getTimeMillis())
+    val corrupt = newLogFile("nonEmptyCorruptLogFile", None, inProgress = false)
+    writeFile(corrupt, true, None, SparkListenerApplicationEnd(0))
+    corrupt.setLastModified(clock.getTimeMillis())
+    logCount += 1
 
     provider.checkForLogs()
     provider.cleanLogs()
-    assert(new File(testDir.toURI).listFiles().size === 3)
+    assert(new File(testDir.toURI).listFiles().size === logCount)
 
     // Move the clock forward 1 day and scan the files again. They should still be there.
     clock.advance(TimeUnit.DAYS.toMillis(1))
     provider.checkForLogs()
     provider.cleanLogs()
-    assert(new File(testDir.toURI).listFiles().size === 3)
+    assert(new File(testDir.toURI).listFiles().size === logCount)
+
+    // Update the slow app to contain valid info. Code should detect the change and not clean
+    // it up.
+    writeFile(slowApp, true, None,
+      SparkListenerApplicationStart(slowApp.getName(), Some(slowApp.getName()), 1L, "test", None))
+    slowApp.setLastModified(clock.getTimeMillis())
+    validLogCount += 1
 
     // Move the clock forward another 2 days and scan the files again. This time the cleaner should
     // pick up the invalid files and get rid of them.
     clock.advance(TimeUnit.DAYS.toMillis(2))
     provider.checkForLogs()
     provider.cleanLogs()
-    assert(new File(testDir.toURI).listFiles().size === 0)
+    assert(new File(testDir.toURI).listFiles().size === validLogCount)
   }
 
   /**