constructed classpath within a session using both the global jars and the session specific jars

Author: wbo4958

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -560,10 +560,15 @@ private[spark] class Executor(
     override def run(): Unit = {
 
       // Classloader isolation
-      val isolatedSession = taskDescription.artifacts.state match {
+      val (isolatedSession, files, jars, archives) = taskDescription.artifacts.state match {
         case Some(jobArtifactState) =>
-          isolatedSessionCache.get(jobArtifactState.uuid, () => newSessionState(jobArtifactState))
-        case _ => defaultSessionState
+          (isolatedSessionCache.get(jobArtifactState.uuid, () => newSessionState(jobArtifactState)),
+            taskDescription.artifacts.files ++ initialUserFiles,
+            taskDescription.artifacts.jars ++ initialUserJars,
+            taskDescription.artifacts.archives ++ initialUserArchives)
+        case _ =>
+          (defaultSessionState, taskDescription.artifacts.files, taskDescription.artifacts.jars,
+          taskDescription.artifacts.archives)
       }
 
       setMDCForTask(taskName, mdcProperties)
@@ -589,11 +594,8 @@ private[spark] class Executor(
         // requires access to properties contained within (e.g. for access control).
         Executor.taskDeserializationProps.set(taskDescription.properties)
 
-        updateDependencies(
-          taskDescription.artifacts.files,
-          taskDescription.artifacts.jars,
-          taskDescription.artifacts.archives,
-          isolatedSession)
+        updateDependencies(files, jars, archives, isolatedSession)
+
         // Always reset the thread class loader to ensure if any updates, all threads (not only
         // the thread that updated the dependencies) can update to the new class loader.
         Thread.currentThread.setContextClassLoader(isolatedSession.replClassLoader)