You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/03/17 13:23:58 UTC
(spark) branch branch-3.5 updated: [SPARK-55998][SHS] Synchronize more places on accessing SHS listing.db
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new b07c654bfa81 [SPARK-55998][SHS] Synchronize more places on accessing SHS listing.db
b07c654bfa81 is described below
commit b07c654bfa81558d39ea68f511bd61354feca99f
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Tue Mar 17 21:12:33 2026 +0800
[SPARK-55998][SHS] Synchronize more places on accessing SHS listing.db
Wrap all
- `listing.delete(classOf[LogInfo], <path>)` and
- `listing.view(classOf[LogInfo])`
operations in SHS's `FsHistoryProvider` with `listing.synchronized { ... }`, this is similar to SPARK-37659.
```
spark.history.fs.cleaner.enabled true
spark.history.fs.cleaner.interval 2h
spark.history.fs.cleaner.maxNum 650000
spark.history.provider org.apache.spark.deploy.history.FsHistoryProvider
spark.history.store.hybridStore.enabled true
spark.history.store.hybridStore.diskBackend ROCKSDB
```
With the above configs, we found that the real preserved number of eventlogs sometimes can exceed 1 million, which reaches the number of files limitation of the HDFS single folder, and then causes the inability to submit new Spark apps (fails due to being unable to create event log file).
After digging the SHS's logs (it's an internal version based on OSS Spark 3.3.4, after taking a look, the master code should have the same issue), we found that in most cases, the scheduled cleanLogs task will fail with
```
2026-02-14 01:05:34 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
java.util.NoSuchElementException: 1^__main__^+hdfs://namenode/spark2-history/application_1770077136288_1231234_1.lz4.inprogress
at org.apache.spark.util.kvstore.RocksDB.get(RocksDB.java:167) ~[spark-kvstore_2.12-3.3.4.113.jar:3.3.4.113]
at org.apache.spark.util.kvstore.RocksDBIterator.next(RocksDBIterator.java:128) ~[spark-kvstore_2.12-3.3.4.113.jar:3.3.4.113]
at scala.collection.convert.Wrappers$JIteratorWrapper.next(Wrappers.scala:46) ~[scala-library-2.12.20.jar:?]
at scala.collection.Iterator.foreach(Iterator.scala:943) ~[scala-library-2.12.20.jar:?]
at scala.collection.Iterator.foreach$(Iterator.scala:943) ~[scala-library-2.12.20.jar:?]
at scala.collection.AbstractIterator.foreach(Iterator.scala:1431) ~[scala-library-2.12.20.jar:?]
at scala.collection.IterableLike.foreach(IterableLike.scala:74) ~[scala-library-2.12.20.jar:?]
at scala.collection.IterableLike.foreach$(IterableLike.scala:73) ~[scala-library-2.12.20.jar:?]
at scala.collection.AbstractIterable.foreach(Iterable.scala:56) ~[scala-library-2.12.20.jar:?]
at scala.collection.TraversableOnce.size(TraversableOnce.scala:139) ~[scala-library-2.12.20.jar:?]
at scala.collection.TraversableOnce.size$(TraversableOnce.scala:129) ~[scala-library-2.12.20.jar:?]
at scala.collection.AbstractTraversable.size(Traversable.scala:108) ~[scala-library-2.12.20.jar:?]
at org.apache.spark.deploy.history.FsHistoryProvider.$anonfun$cleanLogs$1(FsHistoryProvider.scala:1020) ~[spark-core_2.12-3.3.4.113.jar:3.3.4.113]
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) ~[scala-library-2.12.20.jar:?]
at org.apache.spark.util.Utils$.tryLog(Utils.scala:2091) [spark-core_2.12-3.3.4.113.jar:3.3.4.113]
at org.apache.spark.deploy.history.FsHistoryProvider.cleanLogs(FsHistoryProvider.scala:984) [spark-core_2.12-3.3.4.113.jar:3.3.4.113]
at org.apache.spark.deploy.history.FsHistoryProvider.$anonfun$startPolling$4(FsHistoryProvider.scala:326) [spark-core_2.12-3.3.4.113.jar:3.3.4.113]
at org.apache.spark.util.Utils$.tryOrExit(Utils.scala:1436) [spark-core_2.12-3.3.4.113.jar:3.3.4.113]
at org.apache.spark.deploy.history.FsHistoryProvider.$anonfun$getRunner$1(FsHistoryProvider.scala:228) [spark-core_2.12-3.3.4.113.jar:3.3.4.113]
...
```
and the failure causes a significant backlog in the event log folders
```
$ zgrep 'Uncaught exception\|Try to delete' spark-history-server.log.20260214.gz
2026-02-14 01:05:34 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 03:05:40 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 05:07:04 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 07:08:12 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 09:10:02 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 11:11:20 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 13:11:51 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 15:12:33 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 17:13:37 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
2026-02-14 19:14:55 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 55271 old event logs to keep 650000 logs in total.
2026-02-14 21:18:59 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3566 old event logs to keep 650000 logs in total.
2026-02-14 23:19:35 [ERROR] [spark-history-task-0] Utils#302 - Uncaught exception in thread spark-history-task-0
```
Yes, this fixes a bug that SHS may not clean up expired event logs in time.
I have rolled out the patched SHS online for a few days, and am monitoring logs, it has no failures in the cleaning up phase anymore.
```
$ zgrep 'Uncaught exception\|Try to delete' spark-history-server.log.20260314.gz
2026-03-14 01:58:20 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 5886 old event logs to keep 650000 logs in total.
2026-03-14 03:59:49 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 5792 old event logs to keep 650000 logs in total.
2026-03-14 06:00:33 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 6467 old event logs to keep 650000 logs in total.
2026-03-14 08:02:41 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 7531 old event logs to keep 650000 logs in total.
2026-03-14 10:03:39 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 7831 old event logs to keep 650000 logs in total.
2026-03-14 12:06:07 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 5662 old event logs to keep 650000 logs in total.
2026-03-14 14:07:55 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 4894 old event logs to keep 650000 logs in total.
2026-03-14 16:09:04 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 4476 old event logs to keep 650000 logs in total.
2026-03-14 18:11:17 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3757 old event logs to keep 650000 logs in total.
2026-03-14 20:12:05 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3477 old event logs to keep 650000 logs in total.
2026-03-14 22:13:34 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3350 old event logs to keep 650000 logs in total.
$ zgrep 'Uncaught exception\|Try to delete' spark-history-server.log.20260315.gz
2026-03-15 00:15:34 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3480 old event logs to keep 650000 logs in total.
2026-03-15 02:16:24 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 5805 old event logs to keep 650000 logs in total.
2026-03-15 04:17:56 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 6003 old event logs to keep 650000 logs in total.
2026-03-15 06:20:06 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 6576 old event logs to keep 650000 logs in total.
2026-03-15 08:20:52 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 7348 old event logs to keep 650000 logs in total.
2026-03-15 10:24:21 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 7358 old event logs to keep 650000 logs in total.
2026-03-15 12:27:48 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 5537 old event logs to keep 650000 logs in total.
2026-03-15 14:28:35 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 4922 old event logs to keep 650000 logs in total.
2026-03-15 16:29:19 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 4014 old event logs to keep 650000 logs in total.
2026-03-15 18:31:08 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3510 old event logs to keep 650000 logs in total.
2026-03-15 20:33:05 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3351 old event logs to keep 650000 logs in total.
2026-03-15 22:34:55 [INFO] [spark-history-task-0] FsHistoryProvider#185 - Try to delete 3653 old event logs to keep 650000 logs in total.
```
No.
Closes #54817 from pan3793/SPARK-55998.
Authored-by: Cheng Pan <ch...@apache.org>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 2bc6f758fac9ca648b3890579cd44038bee59ee1)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../spark/deploy/history/FsHistoryProvider.scala | 54 +++++++++++++++-------
1 file changed, 38 insertions(+), 16 deletions(-)
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 9841b21861d4..7554bb761891 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
@@ -555,7 +555,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// If the LogInfo read had succeeded, but the ApplicationInafoWrapper
// read failure and throw the exception, we should also cleanup the log
// info from listing db.
- listing.delete(classOf[LogInfo], reader.rootPath.toString)
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], reader.rootPath.toString)
+ }
false
} else if (count < conf.get(UPDATE_BATCHSIZE)) {
listing.write(LogInfo(reader.rootPath.toString(), newLastScanTime,
@@ -606,7 +608,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
.foreach { log =>
log.appId.foreach { appId =>
cleanAppData(appId, log.attemptId, log.logPath)
- listing.delete(classOf[LogInfo], log.logPath)
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], log.logPath)
+ }
}
}
@@ -952,7 +956,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
if (log.lastProcessed <= maxTime && log.appId.isEmpty) {
logInfo(s"Deleting invalid / corrupt event log ${log.logPath}")
deleteLog(fs, new Path(log.logPath))
- listing.delete(classOf[LogInfo], log.logPath)
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], log.logPath)
+ }
}
log.appId.foreach { appId =>
@@ -986,21 +992,27 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
}
// Delete log files that don't have a valid application and exceed the configured max age.
- val stale = KVUtils.viewToSeq(listing.view(classOf[LogInfo])
- .index("lastProcessed")
- .reverse()
- .first(maxTime), Int.MaxValue) { l => l.logType == null || l.logType == LogType.EventLogs }
+ val stale = listing.synchronized {
+ KVUtils.viewToSeq(listing.view(classOf[LogInfo])
+ .index("lastProcessed")
+ .reverse()
+ .first(maxTime), Int.MaxValue) { l => l.logType == null || l.logType == LogType.EventLogs }
+ }
stale.filterNot(isProcessing).foreach { log =>
if (log.appId.isEmpty) {
logInfo(s"Deleting invalid / corrupt event log ${log.logPath}")
deleteLog(fs, new Path(log.logPath))
- listing.delete(classOf[LogInfo], log.logPath)
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], log.logPath)
+ }
}
}
// If the number of files is bigger than MAX_LOG_NUM,
// clean up all completed attempts per application one by one.
- val num = KVUtils.size(listing.view(classOf[LogInfo]).index("lastProcessed"))
+ val num = listing.synchronized {
+ KVUtils.size(listing.view(classOf[LogInfo]).index("lastProcessed"))
+ }
var count = num - maxNum
if (count > 0) {
logInfo(s"Try to delete $count old event logs to keep $maxNum logs in total.")
@@ -1033,7 +1045,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
toDelete.foreach { attempt =>
logInfo(s"Deleting expired event log for ${attempt.logPath}")
val logPath = new Path(logDir, attempt.logPath)
- listing.delete(classOf[LogInfo], logPath.toString())
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], logPath.toString())
+ }
cleanAppData(app.id, attempt.info.attemptId, logPath.toString())
if (deleteLog(fs, logPath)) {
countDeleted += 1
@@ -1081,20 +1095,28 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
}
if (deleteFile) {
logInfo(s"Deleting expired driver log for: ${f.getPath().getName()}")
- listing.delete(classOf[LogInfo], f.getPath().toString())
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], f.getPath().toString())
+ }
deleteLog(driverLogFs, f.getPath())
}
}
// Delete driver log file entries that exceed the configured max age and
// may have been deleted on filesystem externally.
- val stale = KVUtils.viewToSeq(listing.view(classOf[LogInfo])
- .index("lastProcessed")
- .reverse()
- .first(maxTime), Int.MaxValue) { l => l.logType != null && l.logType == LogType.DriverLogs }
+ val stale = listing.synchronized {
+ KVUtils.viewToSeq(listing.view(classOf[LogInfo])
+ .index("lastProcessed")
+ .reverse()
+ .first(maxTime), Int.MaxValue) { l =>
+ l.logType != null && l.logType == LogType.DriverLogs
+ }
+ }
stale.filterNot(isProcessing).foreach { log =>
logInfo(s"Deleting invalid driver log ${log.logPath}")
- listing.delete(classOf[LogInfo], log.logPath)
+ listing.synchronized {
+ listing.delete(classOf[LogInfo], log.logPath)
+ }
deleteLog(driverLogFs, new Path(log.logPath))
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org