sunchao commented on code in PR #59696:
URL: https://github.com/apache/spark/pull/59696#discussion_r4353739195
##########
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:
Verified fixed in e7eb0e10. The live+snapshots probe now persists and
restores the partial metric value 42; live without snapshots keeps its prior
behavior, and both replay controls retain 42. The current-head SQL regressions
also pass. Resolving.
##########
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:
Verified fixed in e7eb0e10. SharedState now supplies the streaming status
listener when snapshots are enabled even if either live-UI setting prevents the
tab, and StreamingQueryManager registers it before events are posted. The new
UI-disabled SharedState test passes in current-head CI. Resolving.
--
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]
