Skip to content

[SPARK-23894][CORE][SQL] Defensively clear ActiveSession in Executors #21185

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

Closed
wants to merge 2 commits into from
Closed

[SPARK-23894][CORE][SQL] Defensively clear ActiveSession in Executors #21185

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2a4944f
[SPARK-23894][CORE][SQL] Defensively clear ActiveSession in Executors
squito Apr 27, 2018
49783c5
review feedback
squito Apr 30, 2018
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
20 changes: 20 additions & 0 deletions core/src/main/scala/org/apache/spark/executor/Executor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,23 @@ private[spark] class Executor(
ManagementFactory.getGarbageCollectorMXBeans.asScala.map(_.getCollectionTime).sum
}

/**
* Only in local mode, we have to prevent the driver from setting the active SparkSession
* in the executor threads. See SPARK-23894.
*/
private lazy val clearActiveSparkSessionMethod = if (Utils.isLocalMaster(conf)) {
Copy link
Member

Choose a reason for hiding this comment

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

Do we still need to call this if we can resolve the root cause in #21190, and issue an error promptly like what you said in #21185 (comment)?

Copy link
Contributor

Choose a reason for hiding this comment

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

I've added this check in #21190

Copy link
Contributor Author

Choose a reason for hiding this comment

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

agreed, I closed this PR

try {
val cls = Utils.classForName("org.apache.spark.sql.SparkSession")
Some(cls.getMethod("clearActiveSession"))
} catch {
case _: ClassNotFoundException =>
// sql not on the classpath, no problem, we don't need to worry about clearing anything
None
}
} else {
None
}

class TaskRunner(
execBackend: ExecutorBackend,
private val taskDescription: TaskDescription)
Expand Down Expand Up @@ -299,6 +316,9 @@ private[spark] class Executor(
Thread.currentThread.setContextClassLoader(replClassLoader)
val ser = env.closureSerializer.newInstance()
logInfo(s"Running $taskName (TID $taskId)")
// When running in local mode, we might end up with the active session from the driver set on
// this thread, though we never should, so we defensively clear it. See SPARK-23894.
clearActiveSparkSessionMethod.foreach(_.invoke(null))
Copy link
Contributor

Choose a reason for hiding this comment

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

Can this be done in the thread pool's thread factory instead?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

ThreadFactories only create threads, they don't start them, so you can't do it directly there, as I need this to run inside the thread.

I thought about trying to do something fancier, like having the thread pool also start the thread and submit one runnable to do this, or having threads get promoted to the correct thread pool after this ran, but I felt like it was too error-prone for minimal gain.

execBackend.statusUpdate(taskId, TaskState.RUNNING, EMPTY_BYTE_BUFFER)
var taskStart: Long = 0
var taskStartCpu: Long = 0
Expand Down