sunchao commented on code in PR #59696:
URL: https://github.com/apache/spark/pull/59696#discussion_r4344405658
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1687,7 +1704,52 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
KVUtils.open(newStorePath, metadata, conf, live = false)
}
- private def createInMemoryStore(attempt: AttemptInfoWrapper): KVStore = {
+ private def createDiskStoreFromSnapshot(
+ dm: HistoryServerDiskManager,
+ appId: String,
+ attempt: AttemptInfoWrapper,
+ metadata: AppStatusStoreMetadata,
+ manifestPath: Path): KVStore = {
+ val lease = dm.leaseExact(snapshotStore.snapshotSize(manifestPath))
+ val startNs = System.nanoTime()
+ val newStorePath = try {
+ Utils.tryWithResource(KVUtils.open(lease.tmpPath, metadata, conf, live =
false)) { store =>
+ snapshotStore.restoreSnapshot(store, manifestPath)
+ }
+ lease.commit(appId, attempt.info.attemptId)
+ } catch {
+ case e: Exception =>
+ lease.rollback()
+ throw e
+ }
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs)
+ logInfo(log"Restored history snapshot ${MDC(PATH, manifestPath)} for " +
+ log"${MDC(APP_ID, appId)} into disk store in ${MDC(TOTAL_TIME,
durationMs)} ms")
+
+ KVUtils.open(newStorePath, metadata, conf, live = false)
+ }
+
+ private def createInMemoryStore(appId: String, attempt: AttemptInfoWrapper):
KVStore = {
+ snapshotStore.findSnapshot(appId, attempt.info.attemptId).foreach {
snapshot =>
+ val store = new InMemoryStore()
+ store.setMetadata(new
AppStatusStoreMetadata(AppStatusStore.CURRENT_VERSION))
+ val startNs = System.nanoTime()
+ try {
+ snapshotStore.restoreSnapshot(store, snapshot)
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ logInfo(log"Restored history snapshot ${MDC(PATH, snapshot)} for " +
+ log"${MDC(APP_ID, appId)} into in-memory store in ${MDC(TOTAL_TIME,
durationMs)} ms")
+ return store
Review Comment:
[P2] Preserve streaming history when the live UI is disabled
For an application with `spark.ui.enabled=false`,
`SharedState.streamingQueryStatusListener` is never installed, although the
core status store and this snapshot are still created. Streaming query events
are still written to the event log, and normal replay installs
`StreamingQueryStatusListener` through the History Server plugin, so those
applications currently have Structured Streaming history. Returning the
snapshot here (and in the disk path) skips that reconstruction: the snapshot
has no streaming rows, and `setupUI()` omits the tab. A focused producer/replay
probe showed `listener=false`, `eventLogged=true`, `liveRows=0`, and
`replayRows=1`. Please collect streaming status when snapshots are enabled
regardless of live-UI settings, or retain replay when that state was not
captured.
##########
core/src/main/scala/org/apache/spark/deploy/history/HistorySnapshotStore.scala:
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.history
+
+import java.io.{BufferedInputStream, BufferedOutputStream, DataInputStream,
DataOutputStream, IOException}
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets
+import java.util.{Properties, UUID}
+
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, Path}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.config.History._
+import org.apache.spark.status._
+import org.apache.spark.status.api.v1
+import org.apache.spark.status.protobuf.KVStoreProtobufSerializer
+import org.apache.spark.util.Utils
+import org.apache.spark.util.kvstore.KVStore
+
+/**
+ * Writes and restores protobuf snapshots of an application's status store
under
+ * `spark.history.snapshot.path`, so the History Server can load a completed
application without
+ * replaying its event log.
+ */
+private[spark] class HistorySnapshotStore(conf: SparkConf, hadoopConf:
Configuration)
+ extends Logging {
+
+ import HistorySnapshotStore._
+
+ def this(conf: SparkConf) = this(conf,
SparkHadoopUtil.get.newConfiguration(conf))
+
+ private val configuredRoot: Option[Path] =
+ if (isEnabled(conf)) conf.get(SNAPSHOT_PATH).map(new Path(_)) else None
+
+ // Resolved lazily so a misconfigured snapshot path does not affect
application startup.
+ private lazy val fs: FileSystem =
configuredRoot.get.getFileSystem(hadoopConf)
+
+ private lazy val rootPath: Option[Path] =
configuredRoot.map(fs.makeQualified)
+
+ private val serializer = new KVStoreProtobufSerializer()
+
+ /**
+ * Writes a versioned history snapshot for an application attempt and
returns the manifest path.
+ *
+ * The snapshot is published under the configured snapshot root using a
fresh manifest and data
+ * directory so readers can safely continue using the previous published
snapshot while a new one
+ * is being written.
+ */
+ def writeSnapshot(store: KVStore, appId: String, attemptId: Option[String]):
Option[Path] = {
+ snapshotDir(appId, attemptId).map { attemptDir =>
+ fs.mkdirs(attemptDir)
+ val snapshotId = newSnapshotId()
+ val dataDir = new Path(attemptDir, s"$SNAPSHOT_DIR_PREFIX$snapshotId")
+ val manifestPath = new Path(attemptDir,
s"$MANIFEST_FILE_PREFIX$snapshotId.properties")
+
+ try {
+ fs.mkdirs(dataDir)
+ val entries = writeEntityFiles(store, dataDir, appId)
+
+ val manifest = new Properties()
+ manifest.setProperty("schemaVersion", SNAPSHOT_SCHEMA_VERSION.toString)
+ manifest.setProperty("snapshotDir", dataDir.getName)
+ manifest.setProperty("classNames",
entries.map(_.className).mkString(","))
+ entries.foreach { entry =>
+ manifest.setProperty(s"${entry.className}.file", entry.fileName)
+ }
+
+ Utils.tryWithResource(fs.create(manifestPath, false)) { out =>
+ manifest.store(out, "spark history snapshot")
+ }
+ // Only older manifests are deleted, so readers always see either the
old snapshot or the
+ // newly published one.
+ cleanupStaleSnapshots(attemptDir, manifestPath)
+ manifestPath
+ } catch {
+ case NonFatal(e) =>
+ deletePathQuietly(manifestPath, recursive = false)
+ deletePathQuietly(dataDir, recursive = true)
+ throw e
+ }
+ }
+ }
+
+ /**
+ * Returns the latest published history snapshot manifest for an application
attempt, if any.
+ *
+ * When multiple manifests exist, the latest versioned manifest is treated
as the active
+ * published snapshot.
+ */
+ def findSnapshot(appId: String, attemptId: Option[String]): Option[Path] = {
+ try {
+ snapshotDir(appId, attemptId).flatMap(latestManifest)
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to look up history snapshot for ${MDC(APP_ID,
appId)}.", e)
+ None
+ }
+ }
+
+ /**
+ * Loads a published history snapshot into the provided KVStore.
+ *
+ * The manifest controls which entity files are restored and where the
snapshot data directory is
+ * located relative to the manifest path.
+ */
+ def restoreSnapshot(store: KVStore, manifestPath: Path): Unit = {
+ val manifest = loadManifest(manifestPath)
+ val schemaVersion = manifest.getProperty("schemaVersion", "-1").toInt
+ require(schemaVersion == SNAPSHOT_SCHEMA_VERSION,
+ s"Unsupported history snapshot version $schemaVersion at $manifestPath")
+
+ val dataRoot = snapshotDataRoot(manifestPath, manifest)
+ manifestEntries(manifestPath, manifest).foreach { entry =>
+ loadSnapshotClass(manifestPath, entry.className).foreach { klass =>
+ readEntityFile(store, klass, new Path(dataRoot, entry.fileName))
+ }
+ }
+ }
+
+ /**
+ * Returns the total on-storage size of a published history snapshot,
including its manifest.
+ *
+ * This is used by History Server disk-store allocation so the target local
store can reserve
+ * enough space before restoring a snapshot.
+ */
+ def snapshotSize(manifestPath: Path): Long = {
+ val manifest = loadManifest(manifestPath)
+ val root = snapshotDataRoot(manifestPath, manifest)
+ manifestEntries(manifestPath, manifest)
+ .foldLeft(fs.getFileStatus(manifestPath).getLen) { (total, entry) =>
+ total + fs.getFileStatus(new Path(root, entry.fileName)).getLen
+ }
+ }
+
+ /**
+ * Deletes all snapshots of an application attempt. The History Server calls
this when it cleans
+ * up the attempt's event log, so snapshots do not outlive their event logs.
+ */
+ def deleteSnapshots(appId: String, attemptId: Option[String]): Unit = {
+ Utils.tryLogNonFatalError {
+ snapshotDir(appId, attemptId).foreach { attemptDir =>
+ deletePathQuietly(attemptDir, recursive = true)
+ val appDir = attemptDir.getParent
+ if (fs.exists(appDir) && fs.listStatus(appDir).isEmpty) {
+ fs.delete(appDir, false)
+ }
+ }
+ }
+ }
+
+ private def snapshotDir(appId: String, attemptId: Option[String]):
Option[Path] = {
+ rootPath.map { root =>
+ new Path(new Path(root, encode(appId)),
encode(attemptId.getOrElse("_default_")))
Review Comment:
[P1] Constrain snapshot cleanup to the intended attempt directory
`URLEncoder` leaves `.` and `..` unchanged, and Hadoop `Path` normalizes
them: an ordinary app ID with attempt ID `..` resolves to the entire snapshot
root, while app ID `..` with attempt ID `.` resolves to its parent. These
identifiers are copied directly from event logs by `AppListingListener`, and
the cleaner calls `deleteSnapshots()` even when that application never wrote a
snapshot. Its recursive delete can therefore remove other applications'
snapshots or unrelated data with the History Server's permissions. Please
reject these components and validate containment before using the path,
especially for deletion. A focused Hadoop Path probe confirmed both
resolutions; the existing disk-cache path's sanitization and canonical-parent
check keep the same inputs contained.
##########
core/src/main/scala/org/apache/spark/deploy/history/HistorySnapshotStore.scala:
##########
@@ -0,0 +1,483 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.deploy.history
+
+import java.io.{BufferedInputStream, BufferedOutputStream, DataInputStream,
DataOutputStream, IOException}
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets
+import java.util.{Properties, UUID}
+
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, Path}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.config.History._
+import org.apache.spark.status._
+import org.apache.spark.status.api.v1
+import org.apache.spark.status.protobuf.KVStoreProtobufSerializer
+import org.apache.spark.util.Utils
+import org.apache.spark.util.kvstore.KVStore
+
+/**
+ * Writes and restores protobuf snapshots of an application's status store
under
+ * `spark.history.snapshot.path`, so the History Server can load a completed
application without
+ * replaying its event log.
+ */
+private[spark] class HistorySnapshotStore(conf: SparkConf, hadoopConf:
Configuration)
+ extends Logging {
+
+ import HistorySnapshotStore._
+
+ def this(conf: SparkConf) = this(conf,
SparkHadoopUtil.get.newConfiguration(conf))
+
+ private val configuredRoot: Option[Path] =
+ if (isEnabled(conf)) conf.get(SNAPSHOT_PATH).map(new Path(_)) else None
+
+ // Resolved lazily so a misconfigured snapshot path does not affect
application startup.
+ private lazy val fs: FileSystem =
configuredRoot.get.getFileSystem(hadoopConf)
+
+ private lazy val rootPath: Option[Path] =
configuredRoot.map(fs.makeQualified)
+
+ private val serializer = new KVStoreProtobufSerializer()
+
+ /**
+ * Writes a versioned history snapshot for an application attempt and
returns the manifest path.
+ *
+ * The snapshot is published under the configured snapshot root using a
fresh manifest and data
+ * directory so readers can safely continue using the previous published
snapshot while a new one
+ * is being written.
+ */
+ def writeSnapshot(store: KVStore, appId: String, attemptId: Option[String]):
Option[Path] = {
+ snapshotDir(appId, attemptId).map { attemptDir =>
+ fs.mkdirs(attemptDir)
+ val snapshotId = newSnapshotId()
+ val dataDir = new Path(attemptDir, s"$SNAPSHOT_DIR_PREFIX$snapshotId")
+ val manifestPath = new Path(attemptDir,
s"$MANIFEST_FILE_PREFIX$snapshotId.properties")
+
+ try {
+ fs.mkdirs(dataDir)
+ val entries = writeEntityFiles(store, dataDir, appId)
+
+ val manifest = new Properties()
+ manifest.setProperty("schemaVersion", SNAPSHOT_SCHEMA_VERSION.toString)
+ manifest.setProperty("snapshotDir", dataDir.getName)
+ manifest.setProperty("classNames",
entries.map(_.className).mkString(","))
+ entries.foreach { entry =>
+ manifest.setProperty(s"${entry.className}.file", entry.fileName)
+ }
+
+ Utils.tryWithResource(fs.create(manifestPath, false)) { out =>
+ manifest.store(out, "spark history snapshot")
+ }
+ // Only older manifests are deleted, so readers always see either the
old snapshot or the
+ // newly published one.
+ cleanupStaleSnapshots(attemptDir, manifestPath)
+ manifestPath
+ } catch {
+ case NonFatal(e) =>
+ deletePathQuietly(manifestPath, recursive = false)
+ deletePathQuietly(dataDir, recursive = true)
+ throw e
+ }
+ }
+ }
+
+ /**
+ * Returns the latest published history snapshot manifest for an application
attempt, if any.
+ *
+ * When multiple manifests exist, the latest versioned manifest is treated
as the active
+ * published snapshot.
+ */
+ def findSnapshot(appId: String, attemptId: Option[String]): Option[Path] = {
+ try {
+ snapshotDir(appId, attemptId).flatMap(latestManifest)
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to look up history snapshot for ${MDC(APP_ID,
appId)}.", e)
+ None
+ }
+ }
+
+ /**
+ * Loads a published history snapshot into the provided KVStore.
+ *
+ * The manifest controls which entity files are restored and where the
snapshot data directory is
+ * located relative to the manifest path.
+ */
+ def restoreSnapshot(store: KVStore, manifestPath: Path): Unit = {
+ val manifest = loadManifest(manifestPath)
+ val schemaVersion = manifest.getProperty("schemaVersion", "-1").toInt
+ require(schemaVersion == SNAPSHOT_SCHEMA_VERSION,
+ s"Unsupported history snapshot version $schemaVersion at $manifestPath")
+
+ val dataRoot = snapshotDataRoot(manifestPath, manifest)
+ manifestEntries(manifestPath, manifest).foreach { entry =>
+ loadSnapshotClass(manifestPath, entry.className).foreach { klass =>
+ readEntityFile(store, klass, new Path(dataRoot, entry.fileName))
+ }
+ }
+ }
+
+ /**
+ * Returns the total on-storage size of a published history snapshot,
including its manifest.
+ *
+ * This is used by History Server disk-store allocation so the target local
store can reserve
+ * enough space before restoring a snapshot.
+ */
+ def snapshotSize(manifestPath: Path): Long = {
+ val manifest = loadManifest(manifestPath)
+ val root = snapshotDataRoot(manifestPath, manifest)
+ manifestEntries(manifestPath, manifest)
+ .foldLeft(fs.getFileStatus(manifestPath).getLen) { (total, entry) =>
+ total + fs.getFileStatus(new Path(root, entry.fileName)).getLen
+ }
+ }
+
+ /**
+ * Deletes all snapshots of an application attempt. The History Server calls
this when it cleans
+ * up the attempt's event log, so snapshots do not outlive their event logs.
+ */
+ def deleteSnapshots(appId: String, attemptId: Option[String]): Unit = {
+ Utils.tryLogNonFatalError {
+ snapshotDir(appId, attemptId).foreach { attemptDir =>
+ deletePathQuietly(attemptDir, recursive = true)
+ val appDir = attemptDir.getParent
+ if (fs.exists(appDir) && fs.listStatus(appDir).isEmpty) {
+ fs.delete(appDir, false)
+ }
+ }
+ }
+ }
+
+ private def snapshotDir(appId: String, attemptId: Option[String]):
Option[Path] = {
+ rootPath.map { root =>
+ new Path(new Path(root, encode(appId)),
encode(attemptId.getOrElse("_default_")))
+ }
+ }
+
+ /** Reads and parses a snapshot manifest file. */
+ private def loadManifest(manifestPath: Path): Properties = {
+ val properties = new Properties()
+ Utils.tryWithResource(fs.open(manifestPath)) { in =>
+ properties.load(in)
+ }
+ properties
+ }
+
+ /** Returns the newest published manifest under a snapshot root, ordered by
manifest name. */
+ private def latestManifest(rootPath: Path): Option[Path] = {
+ if (!fs.exists(rootPath)) {
+ None
+ } else {
+
fs.listStatus(rootPath).map(_.getPath).filter(isVersionedManifest).sortBy(_.getName)
+ .lastOption
+ }
+ }
+
+ /** Removes stale manifests and data only when this writer still owns the
latest publication. */
+ private def cleanupStaleSnapshots(rootPath: Path, keepManifestPath: Path):
Unit = {
+ try {
+ // A concurrent writer may have published a newer manifest after this
write completed.
+ // In that case, avoid deleting anything and let the newer publication
own cleanup.
+ if (latestManifest(rootPath).exists(_ != keepManifestPath)) {
+ logWarning(log"Skipping cleanup for history snapshot root ${MDC(PATH,
rootPath)} " +
+ log"because another manifest became active while publishing " +
+ log"${MDC(PATHS, keepManifestPath)}.")
+ } else {
+ val manifests = fs.listStatus(rootPath).map(_.getPath).toSeq
+ staleManifestsForCleanup(manifests,
keepManifestPath).foreach(deleteSnapshot)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to clean up stale history snapshots under
${MDC(PATH, rootPath)}.",
+ e)
+ }
+ }
+
+ /** Deletes a manifest and its associated snapshot artifacts, tolerating
partial corruption. */
+ private def deleteSnapshot(manifestPath: Path): Unit = {
+ try {
+ deletePathQuietly(snapshotDataRoot(manifestPath,
loadManifest(manifestPath)),
+ recursive = true)
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to load history snapshot manifest at " +
+ log"${MDC(PATH, manifestPath)} for deletion.", e)
+ }
+ deletePathQuietly(manifestPath, recursive = false)
+ }
+
+ private def deletePathQuietly(path: Path, recursive: Boolean): Unit = {
+ Utils.tryLogNonFatalError {
+ if (fs.exists(path)) {
+ fs.delete(path, recursive)
+ }
+ }
+ }
+
+ /** Writes one file per entity type into the snapshot data directory. */
+ private def writeEntityFiles(
+ store: KVStore,
+ dataDir: Path,
+ appId: String): Seq[SnapshotEntry] = {
+ snapshotEntityClasses.zipWithIndex.flatMap { case (klass, idx) =>
+ val fileName = f"$idx%03d-${klass.getSimpleName}.pb"
+ val path = new Path(dataDir, fileName)
+ if (writeEntityFile(store, klass, path, appId) > 0L) {
+ Some(SnapshotEntry(klass.getName, fileName))
+ } else {
+ fs.delete(path, false)
+ None
+ }
+ }
+ }
+
+ /** Writes all records for one entity type into a single snapshot file. */
+ private def writeEntityFile(
+ store: KVStore,
+ klass: Class[_],
+ path: Path,
+ appId: String): Long = {
+ singletonEntity(store, klass, appId)
+ .map { value =>
+ writeValuesFile(path, Iterator.single(redact(value)))
+ }
+ .getOrElse {
+
Utils.tryWithResource(store.view(klass.asInstanceOf[Class[AnyRef]]).closeableIterator())
{
+ it =>
+ writeValuesFile(path,
+
Iterator.continually(it).takeWhile(_.hasNext).map(_.next().asInstanceOf[AnyRef]))
+ }
+ }
+ }
+
+ /** Serializes a stream of KVStore values using length-delimited protobuf
records. */
+ private def writeValuesFile(path: Path, values: Iterator[AnyRef]): Long = {
+ var count = 0L
+ Utils.tryWithResource(new DataOutputStream(new
BufferedOutputStream(fs.create(path, true)))) {
Review Comment:
[P2] Preserve the event log's access restrictions on snapshot files
On a shared filesystem with the default Hadoop umask, these `fs.create`
calls produce `0644` files and the new directories are `0755`. The existing
event-log writer explicitly sets `0660`, consistent with the shared-directory
security guidance. Enabling snapshots therefore makes another copy of
application details, SQL text, and task information readable by filesystem
users who cannot read the corresponding event log. A Hadoop filesystem probe
produced `0644` for the snapshot call versus `0660` for the event-log control.
Please apply the event-log permission policy to the snapshot artifacts before
publishing them; deferring this changes the protection of the same application
data.
##########
core/src/main/scala/org/apache/spark/status/AppStatusListener.scala:
##########
@@ -103,12 +108,18 @@ private[spark] class AppStatusListener(
}
kvstore.onFlush {
- if (!live) {
+ if (!live || historySnapshotStore.isDefined) {
val now = System.nanoTime()
flush(update(_, now))
}
}
+ historySnapshotStore.foreach { snapshotStore =>
+ kvstore.onClose {
+ writeHistorySnapshot(snapshotStore)
Review Comment:
[P2] Flush remaining live SQL metrics before writing the snapshot
The core listener now flushes live state, but `SQLAppStatusListener.onFlush`
still aggregates remaining executions only when `!live`. If the SparkContext
stops while a SQL execution is active, the application-end event can be
recorded before the SQL-execution-end event, and the stopped listener bus drops
that later event. Replay preserves the partial metrics by flushing with
`live=false`; this snapshot instead copies a SQL record with null
`metricValues`, restores successfully, and shows empty metrics. With identical
execution-start, driver-metric-update (`42`), and application-end events, a
focused probe retained `42` after replay but returned an empty metric map after
live-store close. Please flush the SQL listener's remaining state before
publication or fall back to replay for this case.
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1687,7 +1704,52 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
KVUtils.open(newStorePath, metadata, conf, live = false)
}
- private def createInMemoryStore(attempt: AttemptInfoWrapper): KVStore = {
+ private def createDiskStoreFromSnapshot(
+ dm: HistoryServerDiskManager,
+ appId: String,
+ attempt: AttemptInfoWrapper,
+ metadata: AppStatusStoreMetadata,
+ manifestPath: Path): KVStore = {
+ val lease = dm.leaseExact(snapshotStore.snapshotSize(manifestPath))
+ val startNs = System.nanoTime()
+ val newStorePath = try {
+ Utils.tryWithResource(KVUtils.open(lease.tmpPath, metadata, conf, live =
false)) { store =>
+ snapshotStore.restoreSnapshot(store, manifestPath)
+ }
+ lease.commit(appId, attempt.info.attemptId)
Review Comment:
[P2] Validate required application data before accepting the restored store
If the `ApplicationInfoWrapper` entity file is truncated to zero bytes,
`restoreSnapshot()` returns successfully without restoring an application
record. This path then commits the unusable store; the memory path likewise
returns it. `getAppUI()` subsequently constructs `HistoryAppStatusStore`, whose
constructor immediately calls `applicationInfo()` and throws
`NoSuchElementException` outside the snapshot fallback handler. The UI fails
despite a valid event log, and the disk path caches the failure for later
requests. An isolated snapshot round trip confirmed that the valid control
loads, whereas the empty entity is accepted and the next application-info read
fails. Please validate the expected application/attempt before committing or
returning either store, so this corruption triggers replay.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
