zozo123 commented on code in PR #59230: URL: https://github.com/apache/spark/pull/59230#discussion_r4396676606
########## 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: Fixed in 227dfd856d. The suite now skips with an explicit message when the assembly jars directory has no `spark-udf-worker-runtime` jar, so the Maven `sql#core` extended entry no longer fails. The Maven `connect` entry, which rebuilds the assembly, now runs this suite after its own tests, and other entries keep running without an assembly. A GitHub Actions run of those Maven steps on this commit shows the skip and then the pass against the Maven-shaded runtime: https://github.com/zozo123/spark/actions/runs/40623304339 ########## 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: Fixed in 227dfd856d. `SPARK_DIST_CLASSPATH` for the driver now holds only the ScalaTest and Scalactic jars, on top of the assembly, test-classes, and target/jars directories the launcher adds. The suite also fails if `io.grpc.Channel`, `io.perfmark.PerfMark`, `com.google.rpc.Status`, or `com.google.protobuf.Message` load from anywhere outside those launcher jar directories. With the previous classpath this check fails on the first three from the ~/.m2 test jars. One limit remains under sbt. The sbt assembly copies Connect's unshaded gRPC and protobuf jars into the jars directory, so there the check cannot catch a missing relocation. The Maven run covers that, because the Maven assembly ships none of those 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: Fixed in 227dfd856d with the suggested wording. Both builds compile the stubs against `com.google.common`. Maven's inherited module shading relocates those references to `org.sparkproject.guava` and sbt does not, so the runtime relocates both names. ########## 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: Fixed in 227dfd856d. The snapshot, the restore function and its three call sites, and the restore-only try/catch are gone. The overrides are applied to the builder's environment before `start()`. `SparkSubmitSuite` "includes jars passed in through --jars" still passes with no overrides. -- 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]
