From dc6ae098a907690cb297e35f2df07537afe120b5 Mon Sep 17 00:00:00 2001 From: Praj Shastry Date: Thu, 16 Jul 2026 21:22:53 -0400 Subject: [PATCH 1/3] fix(spark4): support Spark 4.2 Jetty 12 (ee10) in history server UI Spark 4.2 upgraded to Jetty 12, shading the servlet classes under org.sparkproject.jetty.ee10.servlet. DataflintJettyUtils only probed the Jetty 9-11 package (org.sparkproject.jetty.servlet / org.eclipse.jetty.servlet), so createStaticHandler failed with ClassNotFoundException: org.eclipse.jetty.servlet.ServletContextHandler and the history UI returned HTTP 500. Extend the reflective class lookup to try the ee10 package first, keeping the older packages as fallbacks (backwards compatible with Spark 3.x / 4.0 / 4.1). Also branch the two Jetty-11-specific init-params that changed in Jetty 12: the DefaultServlet resource param (resourceBase -> baseResource) and the gzip param namespace. Co-Authored-By: Claude Opus 4.8 --- .../dataflint/api/DataflintJettyUtils.scala | 39 ++++++++++++++----- 1 file changed, 29 insertions(+), 10 deletions(-) diff --git a/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala b/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala index 7f7994de..b51f3e20 100644 --- a/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala +++ b/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala @@ -15,25 +15,44 @@ object DataflintJettyUtils { // copy of createStaticHandler in core/src/main/scala/org/apache/spark/ui/JettyUtils.scala // only difference is we are loading the resources from this class loader which might be different from the spark one - // with use reflection to support both org.sparkproject.jetty.servlet and org.eclipse.jetty - // in spark source code + // we use reflection to support the different Jetty package layouts Spark has shipped: + // - org.sparkproject.jetty.ee10.servlet (Spark 4.2+, shaded Jetty 12 / EE10) + // - org.sparkproject.jetty.servlet (Spark 3.x / 4.0 / 4.1, shaded Jetty 9-11) + // - org.eclipse.jetty[.ee10].servlet (unshaded fallbacks) private def createStaticHandler(resourceBase: String, path: String): Any = { - // Try to load classes from both packages + // Try to load the class from every Jetty package layout Spark has used, most-recent first. def getClassForName(className: String): Class[_] = { - try { - Class.forName(s"org.sparkproject.jetty.servlet.$className") - } catch { - case _: ClassNotFoundException => Class.forName(s"org.eclipse.jetty.servlet.$className") - } + val candidates = Seq( + s"org.sparkproject.jetty.ee10.servlet.$className", + s"org.sparkproject.jetty.servlet.$className", + s"org.eclipse.jetty.ee10.servlet.$className", + s"org.eclipse.jetty.servlet.$className" + ) + candidates.iterator + .flatMap { name => + try Some(Class.forName(name)) + catch { case _: ClassNotFoundException => None } + } + .nextOption() + .getOrElse(throw new ClassNotFoundException( + s"Could not load Jetty class '$className' from any known package: ${candidates.mkString(", ")}")) } val servletContextHandlerClass = getClassForName("ServletContextHandler") val defaultServletClass = getClassForName("DefaultServlet") val servletHolderClass = getClassForName("ServletHolder") + // Jetty 12 (EE10) renamed the DefaultServlet resource init-param from "resourceBase" to + // "baseResource" and moved the gzip init-param under the ee10 package namespace. + val isEe10 = servletContextHandlerClass.getName.contains(".ee10.") + val gzipInitParam = + if (isEe10) "org.eclipse.jetty.ee10.servlet.Default.gzip" + else "org.eclipse.jetty.servlet.Default.gzip" + val resourceBaseInitParam = if (isEe10) "baseResource" else "resourceBase" + val contextHandler = servletContextHandlerClass.getDeclaredConstructor().newInstance() val setInitParameterMethod = contextHandler.getClass.getMethod("setInitParameter", classOf[String], classOf[String]) - setInitParameterMethod.invoke(contextHandler, "org.eclipse.jetty.servlet.Default.gzip", "false") + setInitParameterMethod.invoke(contextHandler, gzipInitParam, "false") val staticHandler = defaultServletClass.getDeclaredConstructor().newInstance() val servletHolderConstructor = servletHolderClass.getConstructor(classOf[Servlet]) @@ -42,7 +61,7 @@ object DataflintJettyUtils { Option(this.getClass.getClassLoader.getResource(resourceBase)) match { case Some(res) => val setInitParameterMethodForHolder = holder.getClass.getMethod("setInitParameter", classOf[String], classOf[String]) - setInitParameterMethodForHolder.invoke(holder, "resourceBase", res.toString) + setInitParameterMethodForHolder.invoke(holder, resourceBaseInitParam, res.toString) case None => throw new Exception("Could not find resource path for Web UI: " + resourceBase) } From b94c8c80c6f4afbfff2530b6b982ed9aac50ba9b Mon Sep 17 00:00:00 2001 From: Praj Shastry Date: Thu, 30 Jul 2026 21:53:11 -0400 Subject: [PATCH 2/3] test(spark4): verify Jetty 12 ee10 static UI on Spark 4.2 Add a pluginspark4 integration test against Spark 4.2.0 and an example_4_2_0 project so the Jetty 12 history/live UI fix is covered on the classpath that ships ee10. Co-authored-by: Cursor --- spark-plugin/build.sbt | 24 ++++++-- .../example/ShakespeareSpark420.scala | 42 +++++++++++++ .../api/DataflintJettyUtilsSpec.scala | 60 +++++++++++++++++++ spark-plugin/utils/download-spark-versions.sh | 3 +- 4 files changed, 123 insertions(+), 6 deletions(-) create mode 100644 spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala create mode 100644 spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala diff --git a/spark-plugin/build.sbt b/spark-plugin/build.sbt index 5715a85a..c3552fd4 100644 --- a/spark-plugin/build.sbt +++ b/spark-plugin/build.sbt @@ -20,7 +20,8 @@ lazy val dataflint = project example_3_5_1, example_3_4_1_remote, example_4_0_1, - example_4_1_0 + example_4_1_0, + example_4_2_0 ).settings( crossScalaVersions := Nil, // Aggregate project version must be Nil, see docs: https://www.scala-sbt.org/1.x/docs/Cross-Build.html publish / skip := true @@ -167,13 +168,14 @@ lazy val pluginspark4 = (project in file("pluginspark4")) // Include resources from plugin directory for static UI files Compile / unmanagedResourceDirectories += (plugin / Compile / resourceDirectory).value, - // Test dependencies — Spark 4.0.1 + scalatest. Mirrors pluginspark3 so we can run the + // Test dependencies — Spark 4.2.0 + scalatest. Mirrors pluginspark3 so we can run the // same regression suites against the Spark 4 surface (cross-version validation). + // 4.2.0 ships Jetty 12 (ee10); DataflintJettyUtilsSpec covers that path. // Requires the launching JVM to be Java 17+ since Spark 4 won't run on Java 8/11. libraryDependencies += "org.scalatest" %% "scalatest-funsuite" % "3.2.17" % Test, libraryDependencies += "org.scalatest" %% "scalatest-shouldmatchers" % "3.2.17" % Test, - libraryDependencies += "org.apache.spark" %% "spark-core" % "4.0.1" % Test, - libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.0.1" % Test, + libraryDependencies += "org.apache.spark" %% "spark-core" % "4.2.0" % Test, + libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.2.0" % Test, // Share version-portable test sources with pluginspark3. Most pluginspark3 specs // depend on Spark-3-only internals (Dataset constructor, PythonMapInArrowExec, etc.) @@ -355,4 +357,16 @@ lazy val example_4_1_0 = (project in file("example_4_1_0")) libraryDependencies += "org.apache.spark" % "spark-core_2.13" % "4.1.0", libraryDependencies += "org.apache.spark" % "spark-sql_2.13" % "4.1.0", publish / skip := true - ).dependsOn(pluginspark4) \ No newline at end of file + ).dependsOn(pluginspark4) + +lazy val example_4_2_0 = (project in file("example_4_2_0")) + .settings( + name := "DataflintSparkExample420", + organization := "io.dataflint", + scalaVersion := scala213, + crossScalaVersions := List(scala213), // Only Scala 2.13 for Spark 4.x + // there is no scala 2.12 version so we need to force 2.13 to make it compile + libraryDependencies += "org.apache.spark" % "spark-core_2.13" % "4.2.0", + libraryDependencies += "org.apache.spark" % "spark-sql_2.13" % "4.2.0", + publish / skip := true + ).dependsOn(pluginspark4) diff --git a/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala b/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala new file mode 100644 index 00000000..e32c4b42 --- /dev/null +++ b/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala @@ -0,0 +1,42 @@ +package io.dataflint.example + +import org.apache.spark.sql.{DataFrame, SparkSession} +import org.apache.spark.sql.functions._ + +object ShakespeareSpark420 extends App { + def df(spark: SparkSession): DataFrame = spark.read + .format("csv") + .option("sep", ";") + .option("inferSchema", true) + .load("./test_data/will_play_text.csv") + .toDF("line_id", "play_name", "speech_number", "line_number", "speaker", "text_entry") + .repartition(1000) + + val spark = SparkSession + .builder() + .appName("Shakespeare Statistics") + .config("spark.plugins", "io.dataflint.spark.SparkDataflintPlugin") + .config("spark.dataflint.telemetry.enabled", false) + .config("spark.ui.port", "10000") + .master("local[*]") + .getOrCreate() + + import spark.implicits._ + + val shakespeareText = df(spark) + + shakespeareText.printSchema() + + val count = shakespeareText.count() + println(s"number of records : $count") + + val uniqueSpeakers = shakespeareText.select($"speaker").distinct().count() + println(s"number of unique speakers : $uniqueSpeakers") + + val uniqueWords = shakespeareText.select(explode(split($"text_entry", " "))).distinct().count() + + println(s"number of unique words : $uniqueWords") + + scala.io.StdIn.readLine() + spark.stop() +} diff --git a/spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala b/spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala new file mode 100644 index 00000000..b9a545ca --- /dev/null +++ b/spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala @@ -0,0 +1,60 @@ +package org.apache.spark.dataflint.api + +import java.net.HttpURLConnection +import java.net.URI + +import org.apache.spark.sql.SparkSession +import org.scalatest.BeforeAndAfterAll +import org.scalatest.funsuite.AnyFunSuite +import org.scalatest.matchers.should.Matchers + +/** + * Regression test for Spark 4.2 Jetty 12 (ee10) static-handler support. + * + * Without the ee10 reflective lookup / baseResource init-param fix, + * DataflintJettyUtils.createStaticHandler throws ClassNotFoundException and the + * DataFlint UI returns HTTP 500. + */ +class DataflintJettyUtilsSpec extends AnyFunSuite with Matchers with BeforeAndAfterAll { + + private var spark: SparkSession = _ + + override def beforeAll(): Unit = { + spark = SparkSession.builder() + .master("local[1]") + .appName("DataflintJettyUtilsSpec") + .config("spark.plugins", "io.dataflint.spark.SparkDataflintPlugin") + .config("spark.dataflint.telemetry.enabled", "false") + .config("spark.ui.enabled", "true") + .config("spark.ui.port", "0") + .getOrCreate() + } + + override def afterAll(): Unit = { + if (spark != null) spark.stop() + } + + test("Spark 4.2 classpath exposes Jetty 12 ee10 servlet classes") { + noException should be thrownBy { + Class.forName("org.sparkproject.jetty.ee10.servlet.ServletContextHandler") + Class.forName("org.sparkproject.jetty.ee10.servlet.DefaultServlet") + Class.forName("org.sparkproject.jetty.ee10.servlet.ServletHolder") + } + } + + test("DataFlint static UI handler serves index.html over Jetty 12 ee10") { + val ui = spark.sparkContext.ui.getOrElse( + fail("Spark UI was not started; cannot verify static handler") + ) + val url = s"${ui.webUrl.stripSuffix("/")}/dataflint/index.html" + val connection = URI.create(url).toURL.openConnection().asInstanceOf[HttpURLConnection] + try { + connection.setRequestMethod("GET") + connection.setConnectTimeout(10000) + connection.setReadTimeout(10000) + connection.getResponseCode shouldBe 200 + } finally { + connection.disconnect() + } + } +} diff --git a/spark-plugin/utils/download-spark-versions.sh b/spark-plugin/utils/download-spark-versions.sh index cf1906ca..097b78ec 100755 --- a/spark-plugin/utils/download-spark-versions.sh +++ b/spark-plugin/utils/download-spark-versions.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash # -# Downloads Apache Spark distributions for all supported versions (3.0.x – 4.1.x). +# Downloads Apache Spark distributions for all supported versions (3.0.x – 4.2.x). # Each version is extracted into .spark-versions// under this directory. # # Usage: @@ -26,6 +26,7 @@ ALL_VERSIONS=" 3.5.8:hadoop3 4.0.2:hadoop3 4.1.0:hadoop3 +4.2.0:hadoop3 " get_hadoop_suffix() { From 93b1fe5a1991f53acff6393e6b3381484689757c Mon Sep 17 00:00:00 2001 From: Praj Shastry Date: Thu, 30 Jul 2026 22:19:48 -0400 Subject: [PATCH 3/3] fix(spark4): run Spark 4.2 example on Scala 2.13.18 Spark 4.2 requires Scala 2.13.18+; bump the project Scala version and fork the example run with the build-root CWD so local example_4_2_0/run works. Co-authored-by: Cursor --- spark-plugin/build.sbt | 8 ++++++-- .../io/dataflint/example/ShakespeareSpark420.scala | 13 +++++++++++-- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/spark-plugin/build.sbt b/spark-plugin/build.sbt index c3552fd4..928e543c 100644 --- a/spark-plugin/build.sbt +++ b/spark-plugin/build.sbt @@ -3,7 +3,7 @@ import sbtassembly.AssemblyPlugin.autoImport._ lazy val versionNum: String = "0.9.10" lazy val scala212 = "2.12.20" -lazy val scala213 = "2.13.16" +lazy val scala213 = "2.13.18" lazy val supportedScalaVersions = List(scala212, scala213) lazy val dataflint = project @@ -365,8 +365,12 @@ lazy val example_4_2_0 = (project in file("example_4_2_0")) organization := "io.dataflint", scalaVersion := scala213, crossScalaVersions := List(scala213), // Only Scala 2.13 for Spark 4.x - // there is no scala 2.12 version so we need to force 2.13 to make it compile + // Spark 4.2.0 requires Scala 2.13.18+ (MurmurHash3.caseClassHash) libraryDependencies += "org.apache.spark" % "spark-core_2.13" % "4.2.0", libraryDependencies += "org.apache.spark" % "spark-sql_2.13" % "4.2.0", + // Fork so the run classpath uses this project's Scala, not sbt's. + // Keep CWD at the build root so ./test_data/... resolves like the other examples. + run / fork := true, + run / baseDirectory := (LocalRootProject / baseDirectory).value, publish / skip := true ).dependsOn(pluginspark4) diff --git a/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala b/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala index e32c4b42..abba770d 100644 --- a/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala +++ b/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala @@ -34,9 +34,18 @@ object ShakespeareSpark420 extends App { println(s"number of unique speakers : $uniqueSpeakers") val uniqueWords = shakespeareText.select(explode(split($"text_entry", " "))).distinct().count() - println(s"number of unique words : $uniqueWords") - scala.io.StdIn.readLine() + println("DataFlint UI ready at: http://localhost:10000/dataflint/") + println("Press Enter to stop...") + + // Interactive: wait for Enter. Non-interactive: optional keep-alive for smoke tests. + if (System.console() != null) { + scala.io.StdIn.readLine() + } else { + sys.env.get("DATAFLINT_KEEP_UI_SECONDS").foreach { secs => + Thread.sleep(secs.toLong * 1000) + } + } spark.stop() }