(spark) branch master updated: [SPARK-56278][CORE] Populate accurate metadata immediately during on-demand loading in SHS

[sa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sarutak pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 2addccb6349f [SPARK-56278][CORE] Populate accurate metadata immediately during on-demand loading in SHS
2addccb6349f is described below

commit 2addccb6349fbecc446063fe80cf4e458a9cd0a7
Author: Kousuke Saruta <sa...@amazon.co.jp>
AuthorDate: Mon Mar 30 20:13:02 2026 +0900

    [SPARK-56278][CORE] Populate accurate metadata immediately during on-demand loading in SHS
    
    ### What changes were proposed in this pull request?
    This PR improves the on-demand loading feature in the Spark History Server by populating accurate metadata immediately instead of creating dummy entries.
    
    Previously, when an application was loaded on demand via `getAppUI`, `loadFromFallbackLocation` created a dummy `ApplicationInfoWrapper` with placeholder values (`sparkVersion="unknown"`, `startTime=0`, `completed=false`, etc.) and relied on the next periodic scan cycle to populate accurate information. This meant that the listing would show incorrect metadata until the next scan.
    
    This PR replaces the dummy metadata creation with a direct call to `mergeApplicationListing`, which parses the event log and populates accurate metadata (Spark version, user, start/end time, duration, completion status) immediately. It also removes the `appSparkVersion == "unknown"` cleanup check in `getAppUI` that is no longer needed.
    
    ### Why are the changes needed?
    
    - The listing UI showed `"unknown"` for Spark version and other incorrect metadata for on-demand loaded applications until the next scan cycle
    - Users who increase `spark.history.fs.update.interval` to reduce scan frequency experienced longer periods of inaccurate metadata
    - The dummy metadata path was inconsistent with the scan path, which always uses `mergeApplicationListing`
    
    ### Does this PR introduce _any_ user-facing change?
    Yes. Applications loaded on demand now show accurate metadata (Spark version, user, duration, completion status) immediately in the listing, instead of showing placeholder values until the next scan cycle.
    
    Note: Since `mergeApplicationListing` includes `checkAndCleanLog`, logs that exceed `maxAge` may be cleaned up during on-demand access when `spark.history.fs.cleaner.enabled` is `true`. This is consistent with the behavior during periodic scanning.
    
    ### How was this patch tested?
    Added a new test in `FsHistoryProviderSuite`
    
    ### Was this patch authored or co-authored using generative AI tooling?
    Kiro CLI / Opus 4.6
    
    Closes #55086 from sarutak/shs-accurate-ondemand-metadata.
    
    Authored-by: Kousuke Saruta <sa...@amazon.co.jp>
    Signed-off-by: Kousuke Saruta <sa...@apache.org>
---
 .../spark/deploy/history/FsHistoryProvider.scala   | 38 ++++++++++------------
 .../deploy/history/FsHistoryProviderSuite.scala    | 32 ++++++++++++++++++
 2 files changed, 50 insertions(+), 20 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 54bc290b3787..99d7ca09237d 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
@@ -404,7 +404,10 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
       load(appId)
      } catch {
       case _: NoSuchElementException if this.conf.get(EVENT_LOG_ROLLING_ON_DEMAND_LOAD_ENABLED) =>
-        loadFromFallbackLocation(appId, attemptId, logPath)
+        loadFromFallbackLocation(appId, attemptId, logPath) match {
+          case Some(wrapper) => wrapper
+          case None => return None
+        }
       case _: NoSuchElementException =>
         return None
     }
@@ -426,13 +429,6 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
           createInMemoryStore(attempt)
       }
     } catch {
-      case _: FileNotFoundException if this.conf.get(EVENT_LOG_ROLLING_ON_DEMAND_LOAD_ENABLED) =>
-        if (app.attempts.head.info.appSparkVersion == "unknown") {
-          listing.synchronized {
-            listing.delete(classOf[ApplicationInfoWrapper], appId)
-          }
-        }
-        return None
       case _: FileNotFoundException =>
         return None
     }
@@ -453,18 +449,20 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
   }
 
   private def loadFromFallbackLocation(appId: String, attemptId: Option[String], logPath: String)
-    : ApplicationInfoWrapper = {
-    val date = new Date(0)
-    val lastUpdate = new Date()
-    val (logSourceName, logSourceFullPath) = getLogDirInfo(logPath)
-    val info = ApplicationAttemptInfo(
-      attemptId, date, date, lastUpdate, 0, "spark", false, "unknown",
-      Some(logSourceName), Some(logSourceFullPath))
-    addListing(new ApplicationInfoWrapper(
-      ApplicationInfo(appId, appId, None, None, None, None, List.empty),
-      List(new AttemptInfoWrapper(info, logPath, 0, Some(1), None, None, None, None,
-        logSourceName, logSourceFullPath))))
-    load(appId)
+    : Option[ApplicationInfoWrapper] = {
+    // Call mergeApplicationListing to populate accurate metadata immediately.
+    // logSourceFullPath is empty because on-demand loading has no prior knowledge of the
+    // source directory; resolveLogPath will scan all directories to find the log.
+    val (dirFs, fullPath) = resolveLogPath(logPath, "")
+    try {
+      EventLogFileReader(dirFs, dirFs.getFileStatus(fullPath)).foreach { reader =>
+        mergeApplicationListing(reader, clock.getTimeMillis(), enableOptimizations = true)
+      }
+      Some(load(appId))
+    } catch {
+      case _: FileNotFoundException | _: NoSuchElementException =>
+        None
+    }
   }
 
   override def getEmptyListingHtml(): Seq[Node] = {
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 7c76b50b07ac..8b6076362ee3 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
@@ -1677,6 +1677,38 @@ abstract class FsHistoryProviderSuite extends SparkFunSuite with Matchers with P
     }
   }
 
+  test("SPARK-56278: On-demand loading populates accurate metadata via mergeApplicationListing") {
+    withTempDir { dir =>
+      val conf = createTestConf(true)
+      conf.set(HISTORY_LOG_DIR, dir.getAbsolutePath)
+      conf.set(EVENT_LOG_ROLLING_ON_DEMAND_LOAD_ENABLED, true)
+      val hadoopConf = SparkHadoopUtil.newConfiguration(conf)
+      val provider = new FsHistoryProvider(conf)
+
+      val writer = new RollingEventLogFilesWriter("app1", None, dir.toURI, conf, hadoopConf)
+      writer.start()
+      writeEventsToRollingWriter(writer, Seq(
+        SparkListenerApplicationStart("app1", Some("app1"), 1000, "testuser", None),
+        SparkListenerJobStart(1, 0, Seq.empty),
+        SparkListenerApplicationEnd(5000)), rollFile = false)
+      writer.stop()
+
+      // On-demand load without prior scan
+      assert(provider.getAppUI("app1", None).isDefined)
+
+      // Listing should have accurate metadata from mergeApplicationListing,
+      // not dummy values (sparkVersion="unknown", sparkUser="spark", etc.)
+      val appInfo = provider.getListing().next()
+      assert(appInfo.name === "app1")
+      assert(appInfo.attempts.head.appSparkVersion !== "unknown")
+      assert(appInfo.attempts.head.sparkUser === "testuser")
+      assert(appInfo.attempts.head.completed)
+      assert(appInfo.attempts.head.duration > 0)
+
+      provider.stop()
+    }
+  }
+
   test("SPARK-36354: EventLogFileReader should skip rolling event log directories with no logs") {
     withTempDir { dir =>
       val conf = createTestConf(true)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org