spark-test commented on code in PR #59230: URL: https://github.com/apache/spark/pull/59230#discussion_r4393394361
########## sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/UdfWorkerRuntimeSparkSubmitSuite.scala: ########## @@ -0,0 +1,277 @@ +/* + * 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.execution.externalUDF + +import java.io.File +import java.net.{JarURLConnection, URL} +import java.nio.charset.StandardCharsets +import java.nio.file.Paths +import java.util.jar.JarFile + +import scala.jdk.CollectionConverters._ + +import org.scalatest.Assertions +import org.scalatest.time.SpanSugar._ + +import org.apache.spark.{SparkEnv, TestUtils} +import org.apache.spark.deploy.SparkSubmitTestUtils +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, ExternalUserDefinedFunction} +import org.apache.spark.sql.catalyst.plans.logical.{ExecuteExternalUDF, Range} +import org.apache.spark.sql.internal.StaticSQLConf +import org.apache.spark.sql.types.LongType +import org.apache.spark.tags.ExtendedSQLTest +import org.apache.spark.udf.worker.{DirectWorker, ProcessCallable, UDFProtoCommunicationPattern, + UDFWorkerDataFormat, UDFWorkerProperties, UDFWorkerSpecification, UnixDomainSocket, + WorkerCapabilities, WorkerConnectionSpec, WorkerEnvironment} + +// Spark-submit is required so the driver loads assembly jars plus test-classes. +@ExtendedSQLTest +class UdfWorkerRuntimeSparkSubmitSuite extends SparkSubmitTestUtils { + + import org.apache.spark.udf.worker.grpc.testing.EchoGrpcWorkerMain + + test("packaged UDF worker runtime executes Range(0, 6) with the echo worker") { + assume(EchoGrpcWorkerMain.udsTransportAvailable, + "Netty UDS native transport (epoll on Linux or kqueue on macOS) is required") + val javaBin = Paths.get(System.getProperty("java.home"), "bin", "java").toString + val classPath = System.getProperty("java.class.path") + val workerMain = classOf[EchoGrpcWorkerMain.type].getName.stripSuffix("$") + // spark-internal is SparkLauncher.NO_RESOURCE. A real application jar is added + // to the driver and then fetched from the driver, which deadlocks local mode. + // The spark-submit driver is a new JVM, so it does not inherit the test + // process's module opens. Arrow needs java.nio. The UDS client needs sun.nio.ch. + val driverJavaOptions = Seq( + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED", + "-Dio.netty.tryReflectionSetAccessible=true").mkString(" ") + runSparkSubmit(Seq( + "--class", UdfWorkerRuntimeSparkSubmitSuite.getClass.getName.stripSuffix("$"), + "--master", "local[2]", + "--driver-memory", "1g", + "--conf", "spark.ui.enabled=false", + "--conf", "spark.master.rest.enabled=false", + "--conf", s"spark.driver.extraJavaOptions=$driverJavaOptions", + "spark-internal", + javaBin, + classPath, + workerMain), + timeout = 3.minutes, + environmentOverrides = packagedDriverEnvironment()) + } + + // SPARK_PREPEND_CLASSES puts unshaded module classes ahead of the assembly. + // The dist classpath also carries unshaded udf-worker jars, including a second + // Init.class. Drop those entries and keep the third-party test jars. + private def packagedDriverEnvironment(): Map[String, Option[String]] = { + val dist = Option(new ProcessBuilder().environment().get("SPARK_DIST_CLASSPATH")) + val filtered = dist.map { classpath => + classpath.split(java.io.File.pathSeparator).filterNot(isUnshadedUdfWorker) + .mkString(java.io.File.pathSeparator) + }.filter(_.nonEmpty) + Map( + "SPARK_PREPEND_CLASSES" -> None, + "SPARK_DIST_CLASSPATH" -> filtered) + } + + private def isUnshadedUdfWorker(entry: String): Boolean = { + val path = entry.replace('\\', '/') + path.contains("/udf/worker/") || path.contains("spark-udf-worker-") Review Comment: **Nit (P3):** This filter only removes the unshaded worker modules. Every other entry of sql/core's test classpath stays on the driver classpath, after the assembly jars. Through the test-scoped `spark-udf-worker-grpc` dependency, that includes the gRPC jars, `perfmark-api`, `proto-google-common-protos`, and `protobuf-java`. A distribution ships none of them (see `dev/deps/spark-deps-hadoop-3-hive-2.3`). The jar-layout scan only forbids `io/grpc/`, `com/google/protobuf/`, `com/google/common/`, and `org/sparkproject/guava/`. Suppose a gRPC upgrade adds a transitive dependency that neither the pom `artifactSet` nor the sbt `validPrefixes` lists. Or suppose `perfmark-api` or `proto-google-common-protos` is dropped from the bundle together with its relocation. Either way, the runtime keeps unrelocated references such as `io/perfmark/` or `com/google/rpc/`. A real distribution then fails with `NoClassDefFoundError`. This driver resolves those classes from the test jars, so the layout checks and the `Range(0, 6)` echo plan still pass. The current shading is fine; the guard just cannot catch this class of regression. Could the driver classpath match a distribution instead? Keep the assembly jars and the launcher-provided test-classes directories, and add only the test-support jars the driver's `main` needs (ScalaTest, Scalactic, and similar) in place of the filtered module test classpath. **Verification:** - **Inspection:** Beyond the assembly jars and test-classes directories, the driver classpath the suite constructs contains only the test-support jars its main method needs, with no gRPC, perfmark, protobuf-java, or common-protos jars. - **Behavior:** With the current shading configuration, the packaged suite still loads its main class, passes the jar-layout checks, and returns the Range(0, 6) identity rows using only distribution jars plus those test-support jars. ########## udf/worker/README.md: ########## @@ -78,6 +82,12 @@ generation (from the UDF worker service proto above), and the gRPC runtime depen Keeping gRPC here means `proto/`, `core/`, and their consumers (`core`, `catalyst`, `sql/core`) carry no gRPC dependency on their classpath. +The packaging-only `runtime/` module combines the worker modules for Spark +distributions. It relocates gRPC and protobuf into a UDF-worker namespace while +leaving Netty unrelocated for Spark's existing native transports. Guava is +bundled in the UDF-worker namespace because Maven and SBT compile generated Review Comment: **Nit (P3):** Both builds compile the generated stubs against `com.google.common`, because Guava is a provided-scope dependency. The two package names come from packaging. The root pom's `maven-shade-plugin` execution, which `udf/worker/grpc` inherits, relocates `com.google.common` to `org.sparkproject.guava` in the Maven-built module jar. sbt leaves `com.google.common` alone. That is why `udf/worker/runtime/pom.xml` and `UDFWorkerRuntime` in `SparkBuild.scala` relocate both names. Calling it a compile-time difference sends a maintainer to the compile classpath when they need to decide whether the `org.sparkproject.guava` rule is still required. Possible wording: both builds compile the stubs against `com.google.common`. Maven's inherited module shading relocates those references to `org.sparkproject.guava` in the packaged jars and sbt does not, so the runtime relocates both names and bundles its own Guava. **Verification:** - **Inspection:** The README's Guava explanation matches the root pom's inherited com.google.common relocation and the runtime's relocation of both com.google.common and org.sparkproject.guava. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/UdfWorkerRuntimeSparkSubmitSuite.scala: ########## @@ -0,0 +1,277 @@ +/* + * 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.execution.externalUDF + +import java.io.File +import java.net.{JarURLConnection, URL} +import java.nio.charset.StandardCharsets +import java.nio.file.Paths +import java.util.jar.JarFile + +import scala.jdk.CollectionConverters._ + +import org.scalatest.Assertions +import org.scalatest.time.SpanSugar._ + +import org.apache.spark.{SparkEnv, TestUtils} +import org.apache.spark.deploy.SparkSubmitTestUtils +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, ExternalUserDefinedFunction} +import org.apache.spark.sql.catalyst.plans.logical.{ExecuteExternalUDF, Range} +import org.apache.spark.sql.internal.StaticSQLConf +import org.apache.spark.sql.types.LongType +import org.apache.spark.tags.ExtendedSQLTest +import org.apache.spark.udf.worker.{DirectWorker, ProcessCallable, UDFProtoCommunicationPattern, + UDFWorkerDataFormat, UDFWorkerProperties, UDFWorkerSpecification, UnixDomainSocket, + WorkerCapabilities, WorkerConnectionSpec, WorkerEnvironment} + +// Spark-submit is required so the driver loads assembly jars plus test-classes. +@ExtendedSQLTest +class UdfWorkerRuntimeSparkSubmitSuite extends SparkSubmitTestUtils { + + import org.apache.spark.udf.worker.grpc.testing.EchoGrpcWorkerMain + + test("packaged UDF worker runtime executes Range(0, 6) with the echo worker") { + assume(EchoGrpcWorkerMain.udsTransportAvailable, Review Comment: **Non-blocking (P2):** This `assume` is the suite's only precondition, but the suite also needs a built assembly. `packagedDriverEnvironment()` drops `SPARK_PREPEND_CLASSES` and every udf-worker entry from `SPARK_DIST_CLASSPATH`. After that, the spark-submit child can get the launcher classes, the sql/core classes, and `spark-udf-worker-runtime` only from `assembly/target/scala-2.13/jars`. The scheduled Maven workflows (`build_maven.yml`, `build_maven_java21.yml`, `build_maven_java21_arm.yml`, `build_maven_java21_macos26.yml`, `build_maven_java25.yml`) select this suite through the `sql#core` `ExtendedSQLTest` entry in `maven_test.yml`, and that entry never has an assembly. The precompile artifact excludes `assembly/`, only the connect entry rebuilds it, and the clean-install fallback runs `mvn clean -pl assembly` for every other entry (SPARK-51628). In that job, `bin/spark-class` builds the launch classpath from the missing jars directory alone, so `org.apache.spark.launcher.Main` cannot load and `runSparkSubmit` fails before any assertion runs. Even with a launcher, the driver would have no sql/core main classes and no `Init.class`. The sbt pull-request jobs pass only because they reuse the precompile job's `assembly/package` output. So after merge, that Maven job fails on every run, which also hides new failures in it. The Maven-shaded runtime jar, which Maven-built distributions ship, never gets this suite's checks in CI. This extends an [existing discussion](https://github.com/apache/spark/pull/59230#pullrequestreview-5511649539). **Recommended change:** Make the packaged-runtime suite skip with an explicit message when the assembly jars directory, or the spark-udf-worker-runtime jar inside it, is missing, the same way Connect client tests check for a built assembly. Then run this one suite from a Maven CI step that already has a freshly built assembly, such as the connect entry that rebuilds it, without giving an assembly to the other entries that intentionally run without one. **Why this works:** The guard turns a missing precondition into an explicit skip instead of a launcher failure, so the Maven sql#core extended job stops failing deterministically. The dedicated step gives the Maven-shaded runtime that releases ship the same class-loading, layout, and end-to-end checks the sbt assembly gets today, and the prepend-fallback coverage that the other entries keep by running without an assembly stays intact. **Scope:** Adds an assembly precondition to the packaged-runtime suite and one Maven CI step that runs that suite against the Maven-built assembly. **Compatibility:** The sbt pull-request jobs keep running the packaged-runtime suite against their freshly built assembly, with unchanged assertions. **Risks:** A guard that checks only for the jars directory, and not for the runtime jar inside it, would let a stale assembly without the runtime fail the suite again instead of skipping it. **Constraints:** Keep the other non-connect Maven entries running without an assembly, so spark-submit suites there keep exercising the prepend fallback. **Exclusions:** - Changing how the sbt build_and_test workflow builds or runs the assembly, which already validates the sbt-built runtime on every relevant pull request. **Success:** The scheduled Maven sql#core extended-tests entry no longer fails on the packaged-runtime suite. The suite reports an explicit skip whenever no assembly containing the runtime jar exists. At least one scheduled Maven CI step runs the packaged-runtime suite against a freshly built Maven assembly and fails if the Maven-shaded runtime's layout or end-to-end execution breaks. **Verification:** - **Regression:** Without an assembly jars directory, as in the Maven sql#core extended entry, the packaged-runtime suite skips with an explicit missing-assembly message instead of failing inside spark-submit. - **Behavior:** With a freshly built Maven assembly, the packaged-runtime suite loads Init and the dispatcher factory from the Maven-shaded runtime jar, passes the jar-layout checks, and returns the Range(0, 6) identity rows. - **Compatibility:** The other non-connect Maven entries still run without an assembly, so their spark-submit suites keep exercising the prepend fallback. ########## core/src/test/scala/org/apache/spark/deploy/SparkSubmitTestUtils.scala: ########## @@ -66,13 +67,35 @@ trait SparkSubmitTestUtils extends SparkFunSuite with TimeLimits { env.remove("SPARK_DIST_CLASSPATH") } env.put("SPARK_HOME", sparkHome) + val previousEnvironment = environmentOverrides.keysIterator.map { key => + key -> Option(env.get(key)) + }.toMap + environmentOverrides.foreach { + case (key, Some(value)) => env.put(key, value) + case (key, None) => env.remove(key) + } def captureOutput(source: String)(line: String): Unit = { logInfo(s"$source> $line") history += line } - val process = builder.start() + def restoreEnvironment(): Unit = { + previousEnvironment.foreach { + case (key, Some(value)) => env.put(key, value) + case (key, None) => env.remove(key) + } + } + val process = try { + builder.start() + } catch { + case throwable: Throwable => + restoreEnvironment() + throw throwable + } + // The child copied the overridden environment at start. Put the parent back Review Comment: **Nit (P3):** `env` here is `builder.environment()`, the private copy that this `ProcessBuilder` owns. Per the JDK contract, changing it never affects `System.getenv()` or any other `ProcessBuilder`. `builder` is not used after `start()`, and the next `runSparkSubmit` call starts from a fresh copy of the unchanged JVM environment. So nothing can observe `previousEnvironment`, the `try`/`catch` around `start()`, or the three `restoreEnvironment()` calls. The unrestored `SPARK_TESTING` and `env.remove(...)` writes above already rely on that isolation. This comment states a cleanup contract that does not exist, in a helper several spark-submit suites share, and a later change might rely on it. Applying `environmentOverrides` to `env` before `start()` is enough. You can drop the snapshot, the restore function and its call sites, and the restore-only `try`/`catch`. The comment could instead note that the builder's environment is a per-process copy. **Verification:** - **Behavior:** A spark-submit child launched with overrides sees the overridden values and does not see the removed keys, so the packaged-runtime driver still runs without SPARK_PREPEND_CLASSES and with the filtered distribution classpath. - **Compatibility:** Callers that pass no overrides launch spark-submit with the same environment and the same failure handling as before, including timeout and non-zero exit reporting. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
