Skip to content

[KYUUBI #7625] Avoid resolving the engine result save path during shutdown #7626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wForget wants to merge 4 commits into
base: master
Choose a base branch
from
+17 −13
Open

[KYUUBI #7625] Avoid resolving the engine result save path during shutdown #7626

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b0e8168
[MINOR] Refactor engineSavePath handling in SparkSQLEngine to use Opt…
wForget Aug 12, 2026
9682c77
Apply suggestion
pan3793 Aug 12, 2026
6315834
address comment
wForget Aug 12, 2026
639ad06
fix tests
wForget Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 17 additions & 13 deletions ...yuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import scala.concurrent.duration.Duration
import scala.util.control.NonFatal

import com.google.common.annotations.VisibleForTesting
import org.apache.hadoop.fs.Path
import org.apache.spark.{ui, SparkConf}
import org.apache.spark.kyuubi.{SparkContextHelper, SparkSQLEngineEventListener, SparkSQLEngineListener}
import org.apache.spark.kyuubi.SparkUtilsHelper.getLocalDir
Expand Down Expand Up @@ -58,8 +59,7 @@ case class SparkSQLEngine(spark: SparkSession) extends Serverable("SparkSQLEngin

@volatile private var lifetimeTerminatingChecker: Option[ScheduledExecutorService] = None
@volatile private var stopEngineExec: Option[ThreadPoolExecutor] = None
private lazy val engineSavePath =
backendService.sessionManager.asInstanceOf[SparkSQLSessionManager].getEngineResultSavePath()
@volatile private var engineSavePath: Path = _

@pan3793 pan3793 Aug 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

prefer Option over null

Suggested change
@volatile private var engineSavePath: Path = _
@volatile private var engineSavePath: Option[Path] = None


override def initialize(conf: KyuubiConf): Unit = {
val listener = new SparkSQLEngineListener(this)
Expand Down Expand Up @@ -90,11 +90,13 @@ case class SparkSQLEngine(spark: SparkSession) extends Serverable("SparkSQLEngin
startFastFailChecker(maxInitTimeout)
}

if (backendService.sessionManager.getConf.get(OPERATION_RESULT_SAVE_TO_FILE)) {
val fs = engineSavePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
fs.mkdirs(engineSavePath)
fs.deleteOnExit(engineSavePath)
}
// Due to the fact that session-level enabling of `kyuubi.operation.result.saveToFile.enabled`
// is allowed, we always create and clean up this directory
engineSavePath =
backendService.sessionManager.asInstanceOf[SparkSQLSessionManager].getEngineResultSavePath()
val fs = engineSavePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
fs.mkdirs(engineSavePath)
fs.deleteOnExit(engineSavePath)
Comment on lines +95 to +99

@pan3793 pan3793 Aug 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
engineSavePath =
backendService.sessionManager.asInstanceOf[SparkSQLSessionManager].getEngineResultSavePath()
val fs = engineSavePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
fs.mkdirs(engineSavePath)
fs.deleteOnExit(engineSavePath)
val savePath =
backendService.sessionManager.asInstanceOf[SparkSQLSessionManager].getEngineResultSavePath()
engineSavePath = Some(savePath)
val fs = savePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
fs.mkdirs(savePath)
fs.deleteOnExit(savePath)

}

override def stop(): Unit = if (shutdown.compareAndSet(false, true)) {
Expand All @@ -110,13 +112,15 @@ case class SparkSQLEngine(spark: SparkSession) extends Serverable("SparkSQLEngin
exec,
Duration(60, TimeUnit.SECONDS))
})
try {
val fs = engineSavePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
if (fs.exists(engineSavePath)) {
fs.delete(engineSavePath, true)
if (engineSavePath != null) {
try {
val fs = engineSavePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
if (fs.exists(engineSavePath)) {
fs.delete(engineSavePath, true)
}
} catch {
case e: Throwable => error(s"Error cleaning engine result save path: $engineSavePath", e)
}
} catch {
case e: Throwable => error(s"Error cleaning engine result save path: $engineSavePath", e)
}
Comment on lines +115 to 124

@pan3793 pan3793 Aug 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if (engineSavePath != null) {
try {
val fs = engineSavePath.getFileSystem(spark.sparkContext.hadoopConfiguration)
if (fs.exists(engineSavePath)) {
fs.delete(engineSavePath, true)
}
} catch {
case e: Throwable => error(s"Error cleaning engine result save path: $engineSavePath", e)
}
} catch {
case e: Throwable => error(s"Error cleaning engine result save path: $engineSavePath", e)
}
engineSavePath.foreach { path =>
try {
val fs = path.getFileSystem(spark.sparkContext.hadoopConfiguration)
if (fs.exists(path)) {
fs.delete(path, true)
}
} catch {
case e: Throwable => error(s"Error cleaning engine result save path: $path", e)
}
}

}

Expand Down
Loading