[SPARK-56364][BUILD][TESTS] Generate Scala-based test JARs dynamically instead of storing pre-built binaries

### What changes were proposed in this pull request?
This PR is a part of SPARK-56352 for Scala-based test JARs, replacing pre-built test JAR files containing Scala classes with dynamic compilation at test time, removing 6 binary JAR files and 1 binary from the repository.

Changes:
- Add `TestUtils.createJarWithScalaSources()` in `SparkTestUtils.scala` that compiles Scala source files via `scala.tools.nsc.Main` and packages the resulting classes into a JAR, with support for excluding specific classes by prefix.
- Update test suites to use dynamically generated JARs instead of pre-built ones.
- Refactor `StubClassLoaderSuite` to use a self-contained dummy class instead of the pre-built `udf_noA.jar` that contained spark-connect classes, eliminating the cross-module dependency noted in the original TODO comment.
- Extract `StubClassDummyUdfPacker` from `StubClassDummyUdf.scala` into a separate file for use by `UDFClassLoadingE2ESuite`.
- Remove deleted JAR/binary entries from `dev/test-jars.txt`.

JARs/binaries removed:
- `core/src/test/resources/TestHelloV2_2.13.jar`
- `core/src/test/resources/TestHelloV3_2.13.jar`
- `sql/connect/client/jvm/src/test/resources/TestHelloV2_2.13.jar`
- `sql/connect/client/jvm/src/test/resources/udf2.13.jar`
- `sql/connect/client/jvm/src/test/resources/udf2.13` (serialized binary)
- `sql/core/src/test/resources/artifact-tests/udf_noA.jar`
- `sql/hive/src/test/resources/regression-test-SPARK-8489/test-2.13.jar`

### Why are the changes needed?
As noted in the PR discussion (#50378):
> the ultimate goal is to refactor the tests to automatically build the jars instead of using pre-built ones

This PR achieves that goal for all Scala-based test JARs. By generating JARs dynamically at test time, no binary artifacts need to be stored in the source tree, and the release-time workaround becomes unnecessary for these files.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
All affected test suites pass:
- ClassLoaderIsolationSuite (core)
- StubClassLoaderSuite, ArtifactManagerSuite (sql/core)
- HiveSparkSubmitSuite (sql/hive)
- ReplE2ESuite (sql/connect)
- UDFClassLoadingE2ESuite (sql/connect)

### Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Opus 4.6

Closes #55218 from sarutak/remove-test-jars-c.

Authored-by: Kousuke Saruta <sarutak@amazon.co.jp>
Signed-off-by: Kousuke Saruta <sarutak@apache.org>

Author: sarutak

common/utils/src/main/scala/org/apache/spark/util/SparkTestUtils.scala

@@ -151,6 +151,56 @@ private[spark] trait SparkTestUtils {
     jarFile.toURI.toURL
   }
 
+  /**
+   * Compile Scala source files and package the resulting class files
+   * into a JAR.
+   *
+   * @param sourceFiles Scala source files to compile
+   * @param jarFile the JAR file to create
+   * @param classpathUrls additional classpath URLs needed for compilation
+   * @param excludeClassPrefixes class name prefixes to exclude from the
+   *   JAR (e.g., Seq("A") excludes A.class, A$.class)
+   */
+  def createJarWithScalaSources(
+      sourceFiles: Seq[File],
+      jarFile: File,
+      classpathUrls: Seq[URL] = Seq.empty,
+      excludeClassPrefixes: Seq[String] = Seq.empty): URL = {
+    val classDir = Files.createTempDirectory("spark-test-scala-classes").toFile
+    val compilerClass = SparkClassUtils.classForName[AnyRef]("scala.tools.nsc.Main")
+    val processMethod = compilerClass.getMethod("process", classOf[Array[String]])
+
+    val cpStr = classpathUrls.map(_.getFile).mkString(File.pathSeparator)
+    val args = Array("-classpath", cpStr, "-d", classDir.getAbsolutePath) ++
+      sourceFiles.map(_.getAbsolutePath)
+
+    val success = processMethod.invoke(null, args).asInstanceOf[Boolean]
+    assert(success, s"Scala compilation failed for: ${sourceFiles.map(_.getName).mkString(", ")}")
+
+    try {
+      val classFiles = listFilesRecursively(classDir).filter { f =>
+        f.getName.endsWith(".class") && !excludeClassPrefixes.exists(p =>
+          f.getName == s"$p.class" || f.getName.startsWith(s"$p$$"))
+      }
+
+      jarFile.getParentFile.mkdirs()
+      val jarStream = new JarOutputStream(new FileOutputStream(jarFile))
+      try {
+        for (classFile <- classFiles) {
+          val entryName = classDir.toPath.relativize(classFile.toPath).toString.replace('\\', '/')
+          jarStream.putNextEntry(new JarEntry(entryName))
+          val in = new FileInputStream(classFile)
+          try { in.transferTo(jarStream) } finally { in.close() }
+        }
+      } finally {
+        jarStream.close()
+      }
+    } finally {
+      SparkFileUtils.deleteRecursively(classDir)
+    }
+
+    jarFile.toURI.toURL
+  }
 
   private def listFilesRecursively(dir: File): Seq[File] = {
     val children = dir.listFiles()

core/pom.xml

@@ -39,6 +39,11 @@
       <groupId>org.scala-lang.modules</groupId>
       <artifactId>scala-parallel-collections_${scala.binary.version}</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.scala-lang</groupId>
+      <artifactId>scala-compiler</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.avro</groupId>
       <artifactId>avro</artifactId>

core/src/test/resources/TestHelloV2_2.13.jar

@@ -17,37 +17,38 @@
 
 package org.apache.spark.executor
 
-import java.io.File
+import java.io.{File, PrintWriter}
 import java.net.URL
 
-import scala.util.Properties
-
 import org.apache.spark.{JobArtifactSet, JobArtifactState, LocalSparkContext, SparkConf, SparkContext, SparkFunSuite, TestUtils}
 import org.apache.spark.util.{MutableURLClassLoader, Utils}
 
 
 class ClassLoaderIsolationSuite extends SparkFunSuite with LocalSparkContext  {
 
-  private val scalaVersion = Properties.versionNumberString
-    .split("\\.")
-    .take(2)
-    .mkString(".")
-
   private val jarURL1 = TestUtils.createJarWithClasses(Seq("ClassLoaderIsolation_Dummy"))
   private lazy val jar1 = jarURL1.toString
 
-  // package com.example
-  // object Hello { def test(): Int = 2 }
-  // case class Hello(x: Int, y: Int)
-  private val jarURL2 = Thread.currentThread().getContextClassLoader
-    .getResource(s"TestHelloV2_$scalaVersion.jar")
+  // Dynamically compile Scala source to generate JARs containing
+  // `object Hello { def test(): Int = N }` in package com.example.
+  private def createHelloJar(returnValue: Int): URL = {
+    val source =
+      s"""package com.example
+         |object Hello { def test(): Int = $returnValue }
+         |""".stripMargin
+    val srcFile = File.createTempFile("Hello", ".scala", Utils.createTempDir())
+    val pw = new PrintWriter(srcFile)
+    try { pw.write(source) } finally { pw.close() }
+    val jarFile = File.createTempFile(s"TestHelloV$returnValue", ".jar", Utils.createTempDir())
+    val cp = System.getProperty("java.class.path")
+      .split(File.pathSeparator).map(p => new File(p).toURI.toURL).toSeq
+    TestUtils.createJarWithScalaSources(Seq(srcFile), jarFile, cp)
+  }
+
+  private val jarURL2 = createHelloJar(2)
   private lazy val jar2 = jarURL2.toString
 
-  // package com.example
-  // object Hello { def test(): Int = 3 }
-  // case class Hello(x: String)
-  private val jarURL3 = Thread.currentThread().getContextClassLoader
-    .getResource(s"TestHelloV3_$scalaVersion.jar")
+  private val jarURL3 = createHelloJar(3)
   private lazy val jar3 = jarURL3.toString
 
   test("Executor classloader isolation with JobArtifactSet") {
@@ -122,7 +123,6 @@ class ClassLoaderIsolationSuite extends SparkFunSuite with LocalSparkContext  {
 
   test("SPARK-51537 Executor isolation session classloader inherits from " +
     "default session classloader") {
-    assume(jarURL2 != null)
     sc = new SparkContext(new SparkConf()
       .setAppName("test")
       .setMaster("local")

core/src/test/resources/TestHelloV3_2.13.jar

@@ -1,10 +1,4 @@
-core/src/test/resources/TestHelloV2_2.13.jar
-core/src/test/resources/TestHelloV3_2.13.jar
 data/artifact-tests/junitLargeJar.jar
 data/artifact-tests/smallJar.jar
-sql/connect/client/jvm/src/test/resources/TestHelloV2_2.13.jar
-sql/connect/client/jvm/src/test/resources/udf2.13.jar
 sql/connect/common/src/test/resources/artifact-tests/junitLargeJar.jar
 sql/connect/common/src/test/resources/artifact-tests/smallJar.jar
-sql/core/src/test/resources/artifact-tests/udf_noA.jar
-sql/hive/src/test/resources/regression-test-SPARK-8489/test-2.13.jar

core/src/test/scala/org/apache/spark/executor/ClassLoaderIsolationSuite.scala

@@ -16,41 +16,16 @@
  */
 package org.apache.spark.sql.connect.client
 
-// To generate a jar from the source file:
-// `scalac StubClassDummyUdf.scala -d udf.jar`
-// To remove class A from the jar:
-// `jar -xvf udf.jar` -> delete A.class and A$.class
-// `jar -cvf udf_noA.jar org/`
+/**
+ * Dummy UDF class used for testing dynamic class loading and stub class behavior.
+ *
+ * At test time, this file is compiled by UDFClassLoadingE2ESuite via
+ * createJarWithScalaSources() to generate a JAR and a serialized UdfPacket binary.
+ * StubClassDummyUdfPacker (in a separate file) handles the serialization.
+ */
 class StubClassDummyUdf {
   val udf: Int => Int = (x: Int) => x + 1
   val dummy = (x: Int) => A(x)
 }
 
 case class A(x: Int) { def get: Int = x + 5 }
-
-// The code to generate the udf file
-object StubClassDummyUdf {
-  import java.io.{BufferedOutputStream, File, FileOutputStream}
-  import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.PrimitiveIntEncoder
-  import org.apache.spark.sql.connect.common.UdfPacket
-  import org.apache.spark.util.Utils
-
-  def packDummyUdf(): String = {
-    val byteArray =
-      Utils.serialize[UdfPacket](
-        new UdfPacket(
-          new StubClassDummyUdf().udf,
-          Seq(PrimitiveIntEncoder),
-          PrimitiveIntEncoder
-        )
-      )
-    val file = new File("src/test/resources/udf")
-    val target = new BufferedOutputStream(new FileOutputStream(file))
-    try {
-      target.write(byteArray)
-      file.getAbsolutePath
-    } finally {
-      target.close
-    }
-  }
-}

dev/test-jars.txt

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.client
+
+import java.io.{ByteArrayOutputStream, File, FileOutputStream, ObjectOutputStream}
+
+import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.PrimitiveIntEncoder
+import org.apache.spark.sql.connect.common.UdfPacket
+
+/**
+ * Generates the serialized UdfPacket binary used by UDFClassLoadingE2ESuite.
+ * Run at build time via TestJarTask.buildJarAndRun.
+ */
+object StubClassDummyUdfPacker {
+  def main(args: Array[String]): Unit = {
+    val packet = new UdfPacket(
+      new StubClassDummyUdf().udf,
+      Seq(PrimitiveIntEncoder),
+      PrimitiveIntEncoder)
+    val bos = new ByteArrayOutputStream()
+    val oos = new ObjectOutputStream(bos)
+    oos.writeObject(packet)
+    oos.flush()
+    val fos = new FileOutputStream(new File(args(0)))
+    try { fos.write(bos.toByteArray) } finally { fos.close() }
+  }
+}