This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b3a476d39967 [SPARK-60581][CORE] Delete LogInfo of deleted attempts in
FsHistoryProvider.deleteAttemptLogs
b3a476d39967 is described below
commit b3a476d399678831d990d59d467067624a4ff3d4
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Oct 27 11:05:22 2026 -0700
[SPARK-60581][CORE] Delete LogInfo of deleted attempts in
FsHistoryProvider.deleteAttemptLogs
### What changes were proposed in this pull request?
This PR aims to delete the `LogInfo` listing entries of the deleted
attempts in `FsHistoryProvider.deleteAttemptLogs` again. They were removed
accidentally by SPARK-55793.
The key is the qualified path because `LogInfo` is keyed by
`reader.rootPath`, which is the qualified path returned by
`FileSystem.listStatus`, while `resolveLogPath` returns a path built from the
configured log directory string.
```scala
val (logFs, logPath) = resolveLogPath(attempt.logPath,
attempt.logSourceFullPath)
+ // LogInfo is keyed by the qualified path returned by
FileSystem.listStatus.
+ listing.synchronized {
+ listing.delete(classOf[LogInfo],
logFs.makeQualified(logPath).toString())
+ }
cleanAppData(app.id, attempt.info.attemptId, logPath.toString())
```
### Why are the changes needed?
Since SPARK-55793
([v4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0), 2026-07-11),
`deleteAttemptLogs` deletes the event log files and `ApplicationInfoWrapper`,
but keeps their `LogInfo` entries until the next `checkForLogs` removes them as
stale. Since `cleanLogs` counts `LogInfo` entries for
`spark.history.fs.cleaner.maxNum` right after the max age cleanup in the same
run, the event logs deleted by the max age policy are counted again, and the
cleaner deletes more comple [...]
For example, with `spark.history.fs.cleaner.maxNum=2` and three completed
applications `app1`, `app2`, and `app3` where only `app1` exceeds
`spark.history.fs.cleaner.maxAge`, the cleaner deletes `app1` by the max age
policy, and then deletes `app2` by the max number policy although only two
event logs remain.
In addition, when an event log cannot be deleted, its `LogInfo` entry is
kept and refreshed by every scan. So the application disappears from the
listing permanently and its deletion is not retried.
Note that the original key, `new Path(logDir, attempt.logPath).toString()`,
only matched when `spark.history.fs.logDirectory` was fully-qualified. For a
log directory without a scheme or authority, such as `/path` or `hdfs:///path`
(`new Path("hdfs:///path", "app1")` is `hdfs:/path/app1` while `listStatus`
returns `hdfs://<namenode>/path/app1`), the same over-deletion has existed
since SPARK-28294 added `spark.history.fs.cleaner.maxNum` in
[v3.0.0](https://github.com/apache/spark/rele [...]
### Does this PR introduce _any_ user-facing change?
Yes. The SHS cleaner keeps `spark.history.fs.cleaner.maxNum` event logs
when the max age policy also deletes event logs in the same run.
- For a fully-qualified `spark.history.fs.logDirectory`, this fixes a
regression in Apache Spark 4.2.0 (SPARK-55793).
- For a log directory without a scheme or authority, such as `/path` or
`hdfs:///path`, this fixes a bug that has existed since Apache Spark 3.0.0
(SPARK-28294).
### How was this patch tested?
Pass the CIs with the newly added test case.
```
$ build/sbt 'core/testOnly *RocksDBBackendFsHistoryProviderSuite -- -z
"SPARK-60581"'
```
**BEFORE**
```
[info] - SPARK-60581: log cleaner with both the max age and the maximum
number of log files *** FAILED *** (708 milliseconds)
[info] List("app3") did not contain the same elements as List("app2",
"app3") (FsHistoryProviderSuite.scala:1594)
```
**AFTER**
```
[info] - SPARK-60581: log cleaner with both the max age and the maximum
number of log files (736 milliseconds)
[info] Tests: succeeded 1, failed 0, canceled 0, ignored 0, pending 0
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59833 from dongjoon-hyun/SPARK-60581.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 418bf547488db34c7eedbc96694515b8d1038916)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/deploy/history/FsHistoryProvider.scala | 4 +++
.../deploy/history/FsHistoryProviderSuite.scala | 37 ++++++++++++++++++++++
2 files changed, 41 insertions(+)
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
index 8df2d3e9d724..1c31f49b6378 100644
---
a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -1331,6 +1331,10 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
toDelete.foreach { attempt =>
logInfo(log"Deleting expired event log for ${MDC(PATH,
attempt.logPath)}")
val (logFs, logPath) = resolveLogPath(attempt.logPath,
attempt.logSourceFullPath)
+ // LogInfo is keyed by the qualified path returned by
FileSystem.listStatus.
+ listing.synchronized {
+ listing.delete(classOf[LogInfo],
logFs.makeQualified(logPath).toString())
+ }
cleanAppData(app.id, attempt.info.attemptId, logPath.toString())
if (deleteLog(logFs, logPath)) {
countDeleted += 1
diff --git
a/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
index 0adbae83d144..8fb26e3f1013 100644
---
a/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
+++
b/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
@@ -1485,6 +1485,43 @@ abstract class FsHistoryProviderSuite extends
SparkFunSuite with Matchers with P
}
}
+ test("SPARK-60581: log cleaner with both the max age and the maximum number
of log files") {
+ val maxAge = TimeUnit.SECONDS.toMillis(10)
+ val clock = new ManualClock(maxAge)
+ val conf = createTestConf()
+ .set(CLEANER_ENABLED, true)
+ .set(MAX_LOG_AGE_S.key, s"${maxAge}ms")
+ .set(MAX_LOG_NUM, 2)
+ val provider = new FsHistoryProvider(conf, clock)
+
+ def newCompletedLogFile(appId: String, lastModified: Long): File = {
+ val log = newLogFile(appId, None, inProgress = false)
+ writeFile(log, None,
+ SparkListenerApplicationStart(appId, Some(appId), 1L, "test", None),
+ SparkListenerApplicationEnd(2L))
+ log.setLastModified(lastModified)
+ log
+ }
+ val log1 = newCompletedLogFile("app1", 1000L)
+ val log2 = newCompletedLogFile("app2", 5000L)
+ val log3 = newCompletedLogFile("app3", 6000L)
+
+ // Load all logs while they are within the max age, before the cleaner
runs.
+ provider.checkForLogs()
+ provider.getListing().size should be (3)
+
+ // Only app1 exceeds the max age. Removing it leaves exactly the maximum
number of logs,
+ // so the max number policy should not remove anything more.
+ clock.setTime(maxAge + 2000L)
+ updateAndCheck(provider) { list =>
+ list.map(_.id) should contain theSameElementsAs Seq("app2", "app3")
+ }
+ assert(!log1.exists())
+ assert(log2.exists())
+ assert(log3.exists())
+ assert(provider.listing.count(classOf[LogInfo]) === 2)
+ }
+
test("backwards compatibility with LogInfo from Spark 2.4") {
case class LogInfoV24(
logPath: String,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
