You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2025/04/02 23:43:45 UTC
(spark) branch branch-3.5 updated: [SPARK-51537][CONNECT][CORE][3.5] construct the session-specific classloader based on the default session classloader on executor
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 8eb9e340d12f [SPARK-51537][CONNECT][CORE][3.5] construct the session-specific classloader based on the default session classloader on executor
8eb9e340d12f is described below
commit 8eb9e340d12fa0bd98dea6b87e2a675cb757d3b0
Author: Bobby Wang <wb...@gmail.com>
AuthorDate: Thu Apr 3 08:43:34 2025 +0900
[SPARK-51537][CONNECT][CORE][3.5] construct the session-specific classloader based on the default session classloader on executor
### What changes were proposed in this pull request?
This PR is to construct the session-specific classloader based on the default session classloader which has already added the global jars (e.g., added by `--jars` ) into the classpath on the executor side in the connect mode.
### Why are the changes needed?
In Spark Connect mode, when connecting to a non-local (e.g., standalone) cluster, the executor creates an isolated session state that includes a session-specific classloader for each task. However, a notable issue arises: this session-specific classloader does not include the global JARs specified by the --jars option in the classpath. This oversight can lead to deserialization exceptions. For example:
``` console
Caused by: java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.rdd.MapPartitionsRDD.f of type scala.Function3 in instance of org.apache.spark.rdd.MapPartitionsRDD
at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2096)
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
The newly added test can pass.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #50475 from wbo4958/classloader-3.5.
Authored-by: Bobby Wang <wb...@gmail.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../scala/org/apache/spark/executor/Executor.scala | 43 +++++++++++++++-------
.../spark/executor/ClassLoaderIsolationSuite.scala | 27 ++++++++++++++
2 files changed, 57 insertions(+), 13 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index 69a91839bbeb..03fa0ae6d1fa 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -175,7 +175,8 @@ private[spark] class Executor(
val currentJars = new HashMap[String, Long]
val currentArchives = new HashMap[String, Long]
val urlClassLoader =
- createClassLoader(currentJars, isStubbingEnabledForState(jobArtifactState.uuid))
+ createClassLoader(currentJars, isStubbingEnabledForState(jobArtifactState.uuid),
+ isDefaultState(jobArtifactState.uuid))
val replClassLoader = addReplClassLoaderIfNeeded(
urlClassLoader, jobArtifactState.replClassDirUri, jobArtifactState.uuid)
new IsolatedSessionState(
@@ -1029,7 +1030,8 @@ private[spark] class Executor(
*/
private def createClassLoader(
currentJars: HashMap[String, Long],
- useStub: Boolean): MutableURLClassLoader = {
+ useStub: Boolean,
+ isDefaultSession: Boolean): MutableURLClassLoader = {
// Bootstrap the list of jars with the user class path.
val now = System.currentTimeMillis()
userClassPath.foreach { url =>
@@ -1041,43 +1043,57 @@ private[spark] class Executor(
val urls = userClassPath.toArray ++ currentJars.keySet.map { uri =>
new File(uri.split("/").last).toURI.toURL
}
- createClassLoader(urls, useStub)
+ createClassLoader(urls, useStub, isDefaultSession)
}
- private def createClassLoader(urls: Array[URL], useStub: Boolean): MutableURLClassLoader = {
+ private def createClassLoader(urls: Array[URL],
+ useStub: Boolean,
+ isDefaultSession: Boolean): MutableURLClassLoader = {
logInfo(
s"Starting executor with user classpath (userClassPathFirst = $userClassPathFirst): " +
urls.mkString("'", ",", "'")
)
if (useStub) {
- createClassLoaderWithStub(urls, conf.get(CONNECT_SCALA_UDF_STUB_PREFIXES))
+ createClassLoaderWithStub(urls, conf.get(CONNECT_SCALA_UDF_STUB_PREFIXES), isDefaultSession)
} else {
- createClassLoader(urls)
+ createClassLoader(urls, isDefaultSession)
}
}
- private def createClassLoader(urls: Array[URL]): MutableURLClassLoader = {
+ private def createClassLoader(urls: Array[URL],
+ isDefaultSession: Boolean): MutableURLClassLoader = {
+ // SPARK-51537: The isolated session must *inherit* the classloader from the default session,
+ // which has already included the global JARs specified via --jars. For Spark plugins, we
+ // cannot simply add the plugin JARs to the classpath of the isolated session, as this may
+ // cause the plugin to be reloaded, leading to potential conflicts or unexpected behavior.
+ val loader = if (isDefaultSession) systemLoader else defaultSessionState.replClassLoader
if (userClassPathFirst) {
- new ChildFirstURLClassLoader(urls, systemLoader)
+ new ChildFirstURLClassLoader(urls, loader)
} else {
- new MutableURLClassLoader(urls, systemLoader)
+ new MutableURLClassLoader(urls, loader)
}
}
private def createClassLoaderWithStub(
urls: Array[URL],
- binaryName: Seq[String]): MutableURLClassLoader = {
+ binaryName: Seq[String],
+ isDefaultSession: Boolean): MutableURLClassLoader = {
+ // SPARK-51537: The isolated session must *inherit* the classloader from the default session,
+ // which has already included the global JARs specified via --jars. For Spark plugins, we
+ // cannot simply add the plugin JARs to the classpath of the isolated session, as this may
+ // cause the plugin to be reloaded, leading to potential conflicts or unexpected behavior.
+ val loader = if (isDefaultSession) systemLoader else defaultSessionState.replClassLoader
if (userClassPathFirst) {
// user -> (sys -> stub)
val stubClassLoader =
- StubClassLoader(systemLoader, binaryName)
+ StubClassLoader(loader, binaryName)
new ChildFirstURLClassLoader(urls, stubClassLoader)
} else {
// sys -> user -> stub
val stubClassLoader =
StubClassLoader(null, binaryName)
- new ChildFirstURLClassLoader(urls, stubClassLoader, systemLoader)
+ new ChildFirstURLClassLoader(urls, stubClassLoader, loader)
}
}
@@ -1176,7 +1192,8 @@ private[spark] class Executor(
}
if (renewClassLoader) {
// Recreate the class loader to ensure all classes are updated.
- state.urlClassLoader = createClassLoader(state.urlClassLoader.getURLs, useStub = true)
+ state.urlClassLoader = createClassLoader(state.urlClassLoader.getURLs,
+ useStub = true, isDefaultState(state.sessionUUID))
state.replClassLoader =
addReplClassLoaderIfNeeded(state.urlClassLoader, state.replClassDirUri, state.sessionUUID)
}
diff --git a/core/src/test/scala/org/apache/spark/executor/ClassLoaderIsolationSuite.scala b/core/src/test/scala/org/apache/spark/executor/ClassLoaderIsolationSuite.scala
index 637b459886bc..3f378490a93f 100644
--- a/core/src/test/scala/org/apache/spark/executor/ClassLoaderIsolationSuite.scala
+++ b/core/src/test/scala/org/apache/spark/executor/ClassLoaderIsolationSuite.scala
@@ -109,4 +109,31 @@ class ClassLoaderIsolationSuite extends SparkFunSuite with LocalSparkContext {
}
}
}
+
+ test("SPARK-51537 Executor isolation session classloader inherits from " +
+ "default session classloader") {
+ sc = new SparkContext(new SparkConf()
+ .setAppName("test")
+ .setMaster("local")
+ .set("spark.jars", jar2))
+
+ // TestHelloV2's test method returns '2'
+ val artifactSetWithHelloV2 = new JobArtifactSet(
+ Some(JobArtifactState(uuid = "hello2", replClassDirUri = None)),
+ jars = Map.empty,
+ files = Map.empty,
+ archives = Map.empty
+ )
+
+ JobArtifactSet.withActiveJobArtifactState(artifactSetWithHelloV2.state.get) {
+ sc.parallelize(1 to 1).foreach { i =>
+ val cls = Utils.classForName("com.example.Hello$")
+ val module = cls.getField("MODULE$").get(null)
+ val result = cls.getMethod("test").invoke(module).asInstanceOf[Int]
+ if (result != 2) {
+ throw new RuntimeException("Unexpected result: " + result)
+ }
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org