This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new e1120cd22d1f [SPARK-60580][CORE] Fix NPE in History Server for listing
entries
e1120cd22d1f is described below
commit e1120cd22d1f749f9920bf38ab18837ff30600d1
Author: Kousuke Saruta <[email protected]>
AuthorDate: Wed Oct 28 01:04:02 2026 +0900
[SPARK-60580][CORE] Fix NPE in History Server for listing entries
### What changes were proposed in this pull request?
SPARK-55793 (multiple log directories support for the History Server) added
two plain `String` fields, `logSourceName` and `logSourceFullPath`, to
`AttemptInfoWrapper`, the per-attempt entry stored in the History Server
listing database. It did not change
`FsHistoryProvider.CURRENT_LISTING_VERSION`, so when `spark.history.store.path`
is set and `spark.history.fs.logDirectory` is unchanged, a listing database
written by Spark 4.1 or earlier is kept as is after upgrading to 4.2.0 or lat
[...]
Entries in such a database have neither field. The listing is
(de)serialized as JSON by Jackson in both the `JSON` and `PROTOBUF` store
serializer modes (there is no protobuf SerDe for the history listing classes),
and Jackson fills a missing constructor parameter of type `String` with `null`.
`resolveLogPath` then calls `logSourceFullPath.nonEmpty` and throws a
`NullPointerException`.
This PR makes `resolveLogPath` treat a `null` `logSourceFullPath` like an
empty one, so it falls back to scanning the configured log directories, as it
already does for on-demand loading. Since such an entry can only exist when the
History Server still has the single log directory it had before the upgrade,
the scan finds the right file. `resolveLogPath` is the only place where
`logSourceFullPath` is dereferenced (elsewhere it is only copied into updated
entries), so this covers all o [...]
- `writeEventLogs` (event log download from the UI and the REST API)
- `deleteAttemptLogs` (the log cleaner)
- `createDiskStore` and `createHybridStore` (replaying the event log to
build the application's UI store)
- `createInMemoryStore` (not reachable in this scenario, since the listing
is only persisted together with a local store)
It also documents at the field declarations that the two fields can be
`null`. The `logSourceName` / `logSourceFullPath` fields of
`ApplicationAttemptInfo` are `Option[String]` and are read back as `None`, so
they need no change.
### Why are the changes needed?
When a History Server with `spark.history.store.path` set is upgraded from
Spark 4.1 or earlier to 4.2.0 or later, entries listed before the upgrade keep
the old shape until their event log is parsed again. This covers completed
applications and abandoned in-progress logs, whose files do not change, and,
with `spark.history.fs.inProgressOptimization.enabled` (the default),
applications still running with a single event log file, whose updated entries
copy the old values. For those app [...]
- Downloading the event logs fails with a `NullPointerException`.
- Opening the application UI fails when the application has no usable local
UI store and its event log has to be replayed, for example because it was never
opened, its store was evicted, or (for a running application) the store was
invalidated by a log update.
- The log cleaner stops. In `cleanLogs`, the `NullPointerException` from
the first such expired entry aborts the whole pass. It is caught by
`Utils.tryLog` and logged at ERROR, and every later pass fails again at the
first such expired entry still listed. So deletion of the remaining expired
logs, the cleanup of invalid logs, and the `spark.history.fs.cleaner.maxNum`
limit stop working for the whole History Server.
Bumping `CURRENT_LISTING_VERSION` would also avoid the `null`s, but on the
first start after upgrading, the History Server would delete the whole local
store, including every application's UI store, and re-parse all event logs.
Handling the missing value is cheaper.
This PR does not recover event logs that an unfixed History Server already
left behind. When `deleteAttemptLogs` keeps some attempts of an application
(the unexpired ones, or the incomplete ones for
`spark.history.fs.cleaner.maxNum`), it rewrites the application entry without
the others before deleting their logs. So when it failed with the
`NullPointerException`, those logs stayed on disk with their per-file entry but
are no longer in their application's entry. They are no longer lis [...]
### Does this PR introduce _any_ user-facing change?
Yes, it fixes a bug in Spark 4.2.0 and later. A History Server that reuses
a listing database written by Spark 4.1 or earlier can now download event logs,
load the UI, and clean up the logs of applications listed before the upgrade,
instead of failing with a `NullPointerException`. Until such an entry is
re-parsed, its "Log Source" column in the application list stays empty, the
"Log Source" filter does not match it, and the REST API omits `logSourceName` /
`logSourceFullPath` for it, [...]
### How was this patch tested?
Added a test to `FsHistoryProviderSuite` that rewrites a listing entry
without `logSourceName` / `logSourceFullPath`. The store serializer omits
`null` and `None` fields, so the stored JSON is the same as an entry written by
Spark 4.1. The test checks that the event log can be downloaded, the UI can be
loaded, and the cleaner deletes the expired log and its listing entry. It fails
with the `NullPointerException` without this fix, and passes with it, for both
the `JSON` and `PROTOBUF` [...]
Also checked separately that Jackson reads an entry without the two keys
back with `null` for both fields.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Kiro CLI (auto model selection)
Closes #59830 from sarutak/SPARK-shs-listing-null-logsource.
Authored-by: Kousuke Saruta <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
(cherry picked from commit 1a07005f0c2cf151c9d85963a99910c4ddb8e896)
Signed-off-by: Kousuke Saruta <[email protected]>
---
.../spark/deploy/history/FsHistoryProvider.scala | 11 +++--
.../deploy/history/FsHistoryProviderSuite.scala | 55 ++++++++++++++++++++++
2 files changed, 62 insertions(+), 4 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 3e5c94f4db52..f82f041272e3 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
@@ -859,14 +859,15 @@ private[history] class FsHistoryProvider(conf: SparkConf,
clock: Clock)
* Resolve the log path by searching across all log directories.
* Returns the FileSystem and full Path for the given log path name.
*
- * If logSourceFullPath is available, it is used first to avoid ambiguity
- * when the same filename exists in multiple directories.
+ * If logSourceFullPath is available (non-null and non-empty), it is used
first to avoid
+ * ambiguity when the same filename exists in multiple directories.
* Falls back to scanning all directories if the source path is not found.
*/
private def resolveLogPath(
logPathName: String, logSourceFullPath: String): (FileSystem, Path) = {
- // Try the known source directory first
- if (logSourceFullPath.nonEmpty) {
+ // Try the known source directory first. logSourceFullPath is null for
listing entries
+ // written before SPARK-55793 (Spark 4.1 and earlier), or carried over
from them.
+ if (Option(logSourceFullPath).exists(_.nonEmpty)) {
try {
val sourceFs = logDirFs(logSourceFullPath)
val fullPath = new Path(logSourceFullPath, logPathName)
@@ -1850,6 +1851,8 @@ private[history] class AttemptInfoWrapper(
val viewAcls: Option[String],
val adminAclsGroups: Option[String],
val viewAclsGroups: Option[String],
+ // Both are null for listing entries written before SPARK-55793 (Spark 4.1
and earlier), or
+ // carried over from such entries, which did not record the log source;
see resolveLogPath.
val logSourceName: String,
val logSourceFullPath: String)
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 bba082f02a35..31eb7a67c80e 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
@@ -2436,6 +2436,61 @@ abstract class FsHistoryProviderSuite extends
SparkFunSuite with Matchers with P
}
}
+ test("SPARK-60580: listing entries without a log source from an older
version") {
+ val maxAge = TimeUnit.SECONDS.toMillis(10)
+ val clock = new ManualClock(0)
+ val provider = new FsHistoryProvider(
+ createTestConf().set(MAX_LOG_AGE_S.key, s"${maxAge}ms"), clock)
+
+ val log = newLogFile("app1", None, inProgress = false)
+ writeFile(log, None,
+ SparkListenerApplicationStart("app1", Some("app1-id"), 1L, "test", None),
+ SparkListenerApplicationEnd(2L))
+ log.setLastModified(0L)
+ updateAndCheck(provider) { list =>
+ list.size should be (1)
+ }
+
+ // A listing entry written before SPARK-55793 (Spark 4.1 and earlier) has
no
+ // logSourceName / logSourceFullPath. The wrapper's String fields are read
back as null
+ // and the ApplicationAttemptInfo Options as None. The store serializer
omits null and None
+ // fields (NON_ABSENT), so writing them stores the same JSON as such an
entry.
+ val app = provider.listing.read(classOf[ApplicationInfoWrapper], "app1-id")
+ val oldAttempts = app.attempts.map { a =>
+ new AttemptInfoWrapper(
+ a.info.copy(logSourceName = None, logSourceFullPath = None),
+ a.logPath, a.fileSize, a.lastIndex, a.adminAcls, a.viewAcls,
+ a.adminAclsGroups, a.viewAclsGroups, null, null)
+ }
+ provider.listing.write(new ApplicationInfoWrapper(app.info, oldAttempts))
+ assert(provider.getAttempt("app1-id", None).logSourceFullPath === null)
+
+ // The event log can be downloaded.
+ val bytes = new ByteArrayOutputStream()
+ val zipOut = new ZipOutputStream(bytes)
+ provider.writeEventLogs("app1-id", None, zipOut)
+ zipOut.close()
+ val zipIn = new ZipInputStream(new ByteArrayInputStream(bytes.toByteArray))
+ try {
+ zipIn.getNextEntry.getName should be (log.getName)
+ } finally {
+ zipIn.close()
+ }
+
+ // The UI can be loaded.
+ provider.getAppUI("app1-id", None) should not be None
+
+ // The cleaner deletes the expired log and its listing entry.
+ clock.setTime(maxAge + 1)
+ provider.cleanLogs()
+ assert(!log.exists())
+ intercept[NoSuchElementException] {
+ provider.listing.read(classOf[ApplicationInfoWrapper], "app1-id")
+ }
+
+ provider.stop()
+ }
+
test("SPARK-55864: directory removed while SHS is running") {
val dir2 = Utils.createTempDir(namePrefix = "logDir2")
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
