check and delete expired log when complete process

turboFei · turboFei · commit 832dd7a85ea7 · 2019-11-06T23:49:10.000+08:00
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -682,6 +682,12 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     } finally {
       endProcessing(reader.rootPath)
       pendingReplayTasksCount.decrementAndGet()
+
+      val isExpired = scanTime + conf.get(MAX_LOG_AGE_S) * 1000 > clock.getTimeMillis()
+      if (isExpired) {
+        listing.delete(classOf[LogInfo], reader.rootPath.toString)
+        deleteLog(fs, reader.rootPath)
+      }
     }
   }
 

Original file line number	Diff line number	Diff line change
`@@ -682,6 +682,12 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)`
`682`	`682`	`} finally {`
`683`	`683`	`endProcessing(reader.rootPath)`
`684`	`684`	`pendingReplayTasksCount.decrementAndGet()`
	`685`	`+`
	`686`	`+ val isExpired = scanTime + conf.get(MAX_LOG_AGE_S) * 1000 > clock.getTimeMillis()`
	`687`	`+ if (isExpired) {`
	`688`	`+ listing.delete(classOf[LogInfo], reader.rootPath.toString)`
	`689`	`+ deleteLog(fs, reader.rootPath)`
	`690`	`+ }`
`685`	`691`	`}`
`686`	`692`	`}`
`687`	`693`