[release] fix dataproc failures (#15199)

Fixes the following dataproc test failures in the lastest release
commit:

- VEP serialisation failure
- Warnings from mismatched spark version
- Java/Scala library version conflicts
- Logging configuration errors
- pyarrow abi conflicts in dataproc 2.2.x
- NullPointerException from premature hadoop.fs.closeall call

Major changes:

1. dataproc 2.3.17-debian12

pyarrow had api conflicts with numpy2 in dataproc 2.2.x. dataproc 2.3.x
fixes this
and also provides many of the python libraries we install in
`init_notebook.py`.

1. build.mill

Spark provides many of our runtime dependencies and places its versions
of libraries
ahead of `hail-all-spark.jar` on the classpath. Unless we relocate these
libraries,
we need explicit compile-only dependences on the libraries that we use
directly but
spark ships. Turns out many of spark 3.5.2's 3p libraries are the ones
we need so
don't have to relocate as many!

1. VEP

To reduce what we need to serialise and work around VEP serialisation
failures brought
about by the log4j-api-scala introduction, VEP implementation is now
anonymous. There
are no other functional changes therein.

This change has low security impact on the Broad-managed hail batch
service in GCP,
with changes limited to the version of spark and java logging library
used for query-on-batch.

Author: ehigham

batch/Dockerfile.worker

@@ -46,7 +46,7 @@ RUN hail-pip-install \
     -r hailtop-requirements.txt \
     -r gear-requirements.txt \
     -r batch-requirements.txt \
-    pyspark==3.5.0
+    pyspark==3.5.3
 
 ENV SPARK_HOME=$VIRTUAL_ENV/lib/python$UV_PYTHON/site-packages/pyspark
 ENV PATH="$PATH:$SPARK_HOME/sbin:$SPARK_HOME/bin"

batch/jvm-entryway/build.mill

@@ -20,7 +20,7 @@ object `package` extends ScalafixModule with ScalafmtModule {
 
   override def bomMvnDeps: T[Seq[Dep]] =
     Seq(
-      mvn"org.apache.logging.log4j:log4j-bom:2.25.2",
+      mvn"org.apache.logging.log4j:log4j-bom:2.22.0",
     )
 
   override def scalafixIvyDeps: T[Seq[Dep]] =

hail/Makefile

@@ -20,7 +20,7 @@ HAIL_PIP_VERSION := $(HAIL_MAJOR_MINOR_VERSION).$(HAIL_PATCH_VERSION)
 HAIL_VERSION := $(HAIL_PIP_VERSION)-$(SHORT_REVISION)
 
 SCALA_VERSION ?= 2.12
-SPARK_VERSION ?= 3.5.0
+SPARK_VERSION ?= 3.5.3
 
 include env_var.mk
 

hail/build.mill

@@ -4,14 +4,14 @@
 package build
 
 import com.goyeau.mill.scalafix.ScalafixModule
-import org.typelevel.scalacoptions.*
 import mill.*
 import mill.api.{BuildCtx, Result}
 import mill.scalalib.*
 import mill.scalalib.Assembly.*
 import mill.scalalib.TestModule.TestNg
 import mill.scalalib.scalafmt.ScalafmtModule
 import mill.util.{Jvm, VcsVersion}
+import org.typelevel.scalacoptions.*
 import upickle.default.ReadWriter
 
 object Settings {
@@ -21,16 +21,12 @@ object Settings {
 }
 
 object Deps {
-  object HTTPComponents {
-    val core = mvn"org.apache.httpcomponents:httpcore:4.4.14"
-    val client = mvn"org.apache.httpcomponents:httpclient:4.5.13"
-  }
 
   object Asm {
-    val bom = mvn"org.ow2.asm:asm-bom:9.9"
+    val `asm-bom` = mvn"org.ow2.asm:asm-bom:9.9"
     val core = mvn"org.ow2.asm:asm"
-    val analysis = mvn"org.ow2.asm:asm-analysis"
-    val util = mvn"org.ow2.asm:asm-util"
+    val `asm-analysis` = mvn"org.ow2.asm:asm-analysis"
+    val `asm-util` = mvn"org.ow2.asm:asm-util"
   }
 
   object Breeze {
@@ -39,51 +35,51 @@ object Deps {
     // - https://hail.zulipchat.com/#narrow/stream/123011-Hail-Query-Dev/topic/new.20spark.20ndarray.20failures/near/41645
     // - https://github.com/hail-is/hail/pull/11555
     val core = mvn"org.scalanlp::breeze:1.1"
-    val natives = mvn"org.scalanlp::breeze-natives:1.1"
+    val natives = mvn"org.scalanlp::breeze-natives:1.1".excludeOrg("org.apache.commons.math3")
   }
 
-  object Commons {
-    val io = mvn"commons-io:commons-io:2.20.0"
-    val lang3 = mvn"org.apache.commons:commons-lang3:3.18.0"
-    val codec = mvn"commons-codec:commons-codec:1.19.0"
+  object GoogleCloud {
+    val `libraries-bom` = mvn"com.google.cloud:libraries-bom:26.66.0"
+    val `google-auth-library-oauth2-http` = mvn"com.google.auth:google-auth-library-oauth2-http"
+    val `google-cloud-storage` = mvn"com.google.cloud:google-cloud-storage".excludeOrg("com.fasterxml.jackson.core")
   }
 
   object Spark {
     def core: Task[Dep] = Task.Anon(mvn"org.apache.spark::spark-core:${Env.sparkVersion()}")
     def mllib: Task[Dep] = Task.Anon(mvn"org.apache.spark::spark-mllib:${Env.sparkVersion()}")
-  }
-
-  object Log4j {
-    // have to forceVersion for reasons unknown?
-    val bom = mvn"org.apache.logging.log4j:log4j-bom:2.25.2".forceVersion()
-    val api = mvn"org.apache.logging.log4j:log4j-api"
-    val core = mvn"org.apache.logging.log4j:log4j-core"
-    val slf4j = mvn"org.apache.logging.log4j:log4j-slf4j2-impl"
 
-    val scala = mvn"org.apache.logging.log4j::log4j-api-scala:13.1.0"
+    // direct dependencies provided by Spark at runtime
+    val avro = mvn"org.apache.avro:avro:1.11.2"
+    val `commons-codec` = mvn"commons-codec:commons-codec:1.16.1"
+    val `commons-io` = mvn"commons-io:commons-io:2.16.1"
+    val `commons-lang3` = mvn"org.apache.commons:commons-lang3:3.12.0"
+    val `commons-math3` = mvn"org.apache.commons:commons-math3:3.6.1"
+    val `hadoop-client-api` = mvn"org.apache.hadoop:hadoop-client-api:3.3.4"
+    val httpclient = mvn"org.apache.httpcomponents:httpclient:4.5.14"
+    val `jackson-core` = mvn"com.fasterxml.jackson.core:jackson-core:2.15.2"
+    val `json4s-jackson` = mvn"org.json4s::json4s-jackson:3.7.0-M11"
+    val `l4z-java` = mvn"org.lz4:lz4-java:1.8.0"
+    val `log4j-bom` = mvn"org.apache.logging.log4j:log4j-bom:2.22.0"
+    val `logj4-api` = mvn"org.apache.logging.log4j:log4j-api"
+    val `log4j-core` = mvn"org.apache.logging.log4j:log4j-core"
+    val `zstd-jni` = mvn"com.github.luben:zstd-jni:1.5.5-4"
   }
 
-  val samtools = mvn"com.github.samtools:htsjdk:3.0.5"
-  val jdistlib = mvn"net.sourceforge.jdistlib:jdistlib:0.4.5"
+  val `elasticsearch-spark` = mvn"org.elasticsearch::elasticsearch-spark-30:8.4.3".excludeOrg("org.apache.spark")
   val freemarker = mvn"org.freemarker:freemarker:2.3.31"
-  val elasticsearch = mvn"org.elasticsearch::elasticsearch-spark-30:8.4.3"
-  val gcloud = mvn"com.google.cloud:google-cloud-storage:2.55.0"
+  val htsjdk = mvn"com.github.samtools:htsjdk:3.0.5".excludeOrg("*")
+  val jdistlib = mvn"net.sourceforge.jdistlib:jdistlib:0.4.5".excludeOrg("*")
   val jna = mvn"net.java.dev.jna:jna:5.13.0"
-  val json4s = mvn"org.json4s::json4s-jackson:3.7.0-M11"
-  val zstd = mvn"com.github.luben:zstd-jni:1.5.5-11"
-  val lz4 = mvn"org.lz4:lz4-java:1.8.0"
+  val `junixsocket-core` = mvn"com.kohlschutter.junixsocket:junixsocket-core:2.6.1"
+  val `log4j-api-scala` = mvn"org.apache.logging.log4j::log4j-api-scala:13.1.0".excludeOrg("*")
   val netlib = mvn"com.github.fommil.netlib:all:1.1.2"
-  val avro = mvn"org.apache.avro:avro:1.11.2"
-  val junixsocket = mvn"com.kohlschutter.junixsocket:junixsocket-core:2.6.1"
-  val hadoopClient = mvn"org.apache.hadoop:hadoop-client:3.4.2"
-  val jackson = mvn"com.fasterxml.jackson.core:jackson-core:2.15.2"
-  val sourcecode = mvn"com.lihaoyi::sourcecode:0.4.2"
-  val collection_compat = mvn"org.scala-lang.modules::scala-collection-compat:2.13.0"
+  val `scala-collection-compat` = mvn"org.scala-lang.modules::scala-collection-compat:2.13.0"
   // provides @nowarn212 and @nowarn213
-  val scalac_compat = mvn"org.typelevel::scalac-compat-annotation:0.1.4"
+  val `scalac-compat-annotation` = mvn"org.typelevel::scalac-compat-annotation:0.1.4"
+  val sourcecode = mvn"com.lihaoyi::sourcecode:0.4.2"
 
   object Plugins {
-    val betterModadicFor = mvn"com.olegpy::better-monadic-for:0.3.1"
+    val `better-monadic-for` = mvn"com.olegpy::better-monadic-for:0.3.1"
   }
 }
 
@@ -97,7 +93,7 @@ object Env extends Module {
   }
 
   def sparkVersion: T[String] = Task.Input {
-    Result.Success(Task.env.getOrElse("SPARK_VERSION", "3.5.0"))
+    Result.Success(Task.env.getOrElse("SPARK_VERSION", "3.5.3"))
   }
 
   val buildMode: BuildMode = BuildMode.valueOf(millbuild.BuildConfig.buildMode)
@@ -113,9 +109,10 @@ trait HailModule extends ScalaModule with ScalafmtModule with ScalafixModule { o
     mvn"io.github.tanin47::scalafix-forbidden-symbol:1.0.0",
   )
 
-  override def depManagement: T[Seq[Dep]] = Task {
-    Seq(Deps.collection_compat)
-  }
+  override def depManagement: T[Seq[Dep]] =
+    Seq(
+      Deps.`scala-collection-compat`,
+    )
 
   override def javacOptions: T[Seq[String]] = Seq(
     "-Xlint:all",
@@ -215,24 +212,6 @@ trait HailModule extends ScalaModule with ScalafmtModule with ScalafixModule { o
       finalOptions :+ "-Wconf:msg=legacy-binding:s"
     }
   }
-
-  trait HailTests extends ScalaTests with TestNg with ScalafmtModule with ScalafixModule {
-    override def forkArgs: T[Seq[String]] = Seq("-Xss4m", "-Xmx4096M")
-
-    override def scalafixIvyDeps: T[Seq[Dep]] = Seq(
-      mvn"io.github.tanin47::scalafix-forbidden-symbol:1.0.0",
-    )
-
-    override def mvnDeps: T[Seq[Dep]] =
-      outer.compileMvnDeps() ++ Seq(
-        mvn"org.scalacheck::scalacheck:1.18.1",
-        mvn"org.scalatest::scalatest:3.2.19",
-        mvn"org.scalatest::scalatest-shouldmatchers:3.2.19",
-        mvn"org.scalatestplus::scalacheck-1-18:3.2.19.0",
-        mvn"org.scalatestplus::testng-7-10:3.2.19.0",
-        mvn"org.mockito::mockito-scala:1.17.31",
-      )
-  }
 }
 
 object hail extends Cross[RootHailModule](millbuild.BuildConfig.enabledScalaVersions)
@@ -276,70 +255,64 @@ trait RootHailModule extends CrossScalaModule with HailModule { outer =>
   }
 
   override def bomMvnDeps: T[Seq[Dep]] = Seq(
-    Deps.Asm.bom,
-    Deps.Log4j.bom,
+    Deps.Asm.`asm-bom`,
+    Deps.GoogleCloud.`libraries-bom`,
+    Deps.Spark.`log4j-bom`,
   )
 
   override def mvnDeps: T[Seq[Dep]] = Seq(
-    Deps.HTTPComponents.core,
-    Deps.HTTPComponents.client,
     Deps.Asm.core,
-    Deps.Asm.analysis,
-    Deps.Asm.util,
-    Deps.Log4j.core,
-    Deps.Log4j.scala,
-    Deps.samtools.excludeOrg("*"),
-    Deps.jdistlib.excludeOrg("*"),
+    Deps.Asm.`asm-analysis`,
+    Deps.Asm.`asm-util`,
+    Deps.GoogleCloud.`google-auth-library-oauth2-http`,
+    Deps.GoogleCloud.`google-cloud-storage`,
+    Deps.`elasticsearch-spark`,
     Deps.freemarker,
-    Deps.elasticsearch.excludeOrg("org.apache.spark"),
-    Deps.gcloud.excludeOrg("com.fasterxml.jackson.core"),
+    Deps.htsjdk,
+    Deps.jdistlib,
+    Deps.`log4j-api-scala`,
     Deps.jna,
-    Deps.json4s.excludeOrg("com.fasterxml.jackson.core"),
-    Deps.zstd,
+    Deps.`scala-collection-compat`,
     Deps.sourcecode,
-    Deps.collection_compat,
   )
 
   override def runMvnDeps: T[Seq[Dep]] = Seq(
-    Deps.Breeze.natives.excludeOrg("org.apache.commons.math3"),
-    Deps.Commons.io,
-    Deps.Commons.lang3,
-    Deps.Commons.codec,
-    Deps.lz4,
+    Deps.Breeze.natives,
     Deps.netlib,
-    Deps.avro.excludeOrg("com.fasterxml.jackson.core"),
-    Deps.junixsocket,
-    Deps.Log4j.slf4j,
+    Deps.`junixsocket-core`,
   )
 
   override def compileMvnDeps: T[Seq[Dep]] = Seq(
-    Deps.Log4j.api,
-    Deps.hadoopClient,
+    Deps.Breeze.core,
     Deps.Spark.core().excludeOrg("org.scalanlp"),  // Hail has an explicit dependency on Breeze 1.1
     Deps.Spark.mllib().excludeOrg("org.scalanlp"),  // Hail has an explicit dependency on Breeze 1.1
-    Deps.Breeze.core,
-    Deps.jackson,
-    Deps.scalac_compat,
+    Deps.Spark.avro,
+    Deps.Spark.`commons-codec`,
+    Deps.Spark.`commons-io`,
+    Deps.Spark.`commons-lang3`,
+    Deps.Spark.`commons-math3`,
+    Deps.Spark.httpclient,
+    Deps.Spark.`hadoop-client-api`,
+    Deps.Spark.`jackson-core`,
+    Deps.Spark.`json4s-jackson`,
+    Deps.Spark.`l4z-java`,
+    Deps.Spark.`logj4-api`,
+    Deps.Spark.`log4j-core`,
+    Deps.Spark.`zstd-jni`,
+    Deps.`scalac-compat-annotation`,
   )
 
   override def assemblyRules: Seq[Rule] = super.assemblyRules ++ Seq(
     Rule.Exclude("META-INF/INDEX.LIST"),
     Rule.ExcludePattern("^scala/(?!collection/compat).*"),
     Rule.AppendPattern("META-INF/services/.*", "\n"),
     Rule.Relocate("breeze.**", "is.hail.relocated.@0"),
-    Rule.Relocate("com.google.cloud.**", "is.hail.relocated.@0"),
-    Rule.Relocate("com.google.common.**", "is.hail.relocated.@0"),
-    Rule.Relocate("org.apache.commons.io.**", "is.hail.relocated.@0"),
-    Rule.Relocate("org.apache.commons.lang3.**", "is.hail.relocated.@0"),
-    Rule.Relocate("org.apache.http.**", "is.hail.relocated.@0"),
-    Rule.Relocate("org.elasticsearch.**", "is.hail.relocated.@0"),
-    Rule.Relocate("org.json4s.**", "is.hail.relocated.@0"),
-    Rule.Relocate("org.objectweb.**", "is.hail.relocated.@0"),
+    Rule.Relocate("com.google.**", "is.hail.relocated.@0"),
     Rule.Relocate("scala.collection.compat.**", "is.hail.relocated.@0"),
   )
 
   override def scalacPluginMvnDeps: T[Seq[Dep]] = Seq(
-    Deps.Plugins.betterModadicFor
+    Deps.Plugins.`better-monadic-for`,
   )
 
   def writeRunClasspath: T[PathRef] = Task {
@@ -383,18 +356,32 @@ trait RootHailModule extends CrossScalaModule with HailModule { outer =>
     }
   }
 
-  object test extends HailTests {
-    override def assemblyRules: Seq[Rule] = outer.assemblyRules ++ Seq(
-      Rule.Relocate("org.codehaus.jackson.**", "is.hail.relocated.@0")
-    )
+  object test extends ScalaTests with TestNg with ScalafmtModule with ScalafixModule {
+    override def forkArgs: T[Seq[String]] =
+      Seq("-Xss4m", "-Xmx4096M")
+
+    override def scalafixIvyDeps: T[Seq[Dep]] =
+      outer.scalafixIvyDeps()
+
+    override def compileMvnDeps: T[Seq[Dep]] =
+       outer.compileMvnDeps()
 
     override def mvnDeps: T[Seq[Dep]] =
-      super.mvnDeps() ++ outer.mvnDeps() ++ Seq(
-        Deps.jackson,
+      outer.mvnDeps() ++ Seq(
+        mvn"com.google.inject:guice:5.1.0",
+        mvn"org.scalacheck::scalacheck:1.18.1",
+        mvn"org.scalatest::scalatest:3.2.19",
+        mvn"org.scalatest::scalatest-shouldmatchers:3.2.19",
+        mvn"org.scalatestplus::scalacheck-1-18:3.2.19.0",
+        mvn"org.scalatestplus::testng-7-10:3.2.19.0",
+        mvn"org.mockito::mockito-scala:1.17.31",
       )
 
-    override def compileModuleDeps: Seq[JavaModule] =
-      super.compileModuleDeps ++ outer.compileModuleDeps
+    override def runMvnDeps: T[Seq[Dep]] =
+      outer.runMvnDeps()
+
+    override def assemblyRules: Seq[Rule] =
+      outer.assemblyRules
   }
 
   object shadedazure extends JavaModule {

hail/hail/src/is/hail/backend/driver/Py4JQueryDriver.scala

@@ -28,7 +28,6 @@ import java.io.Closeable
 import java.net.InetSocketAddress
 import java.util
 
-import com.google.api.client.http.HttpStatusCodes
 import com.sun.net.httpserver.{HttpExchange, HttpServer}
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.sql.DataFrame
@@ -412,12 +411,12 @@ final class Py4JQueryDriver(backend: Backend) extends Closeable with Logging {
         }
 
         override def result(req: HttpExchange, result: Array[Byte]): Unit =
-          respond(req, HttpStatusCodes.STATUS_CODE_OK, result)
+          respond(req, 200, result)
 
         override def failure(req: HttpExchange, t: Throwable): Unit =
           respond(
             req,
-            HttpStatusCodes.STATUS_CODE_SERVER_ERROR,
+            500,
             jsonToBytes {
               val (shortMessage, expandedMessage, errorId) = handleForPython(t)
               JObject(

hail/hail/src/is/hail/backend/service/Worker.scala

@@ -17,8 +17,7 @@ import java.nio.charset._
 import java.nio.file.Path
 import java.util
 import java.util.concurrent.Executors
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder
+import java.util.concurrent.atomic.AtomicInteger
 
 class ServiceTaskContext(val partitionId: Int) extends HailTaskContext {
   override def stageId(): Int = 0
@@ -193,12 +192,15 @@ object Worker extends Logging {
 
     implicit val ec: ExecutionContext =
       ExecutionContext.fromExecutor(
-        Executors.newCachedThreadPool(
-          new ThreadFactoryBuilder()
-            .setDaemon(true)
-            .setNameFormat("hail-worker-thread-%d")
-            .build()
-        )
+        Executors.newCachedThreadPool {
+          val threadFactory = Executors.defaultThreadFactory()
+          val counter = new AtomicInteger(0)
+          task =>
+            val thread = threadFactory.newThread(task)
+            thread.setName(f"hail-worker-thread-${counter.getAndIncrement()}")
+            thread.setDaemon(true)
+            thread
+        }
       )
 
     def open(x: String): SeekableDataInputStream =