haiyangsun-db commented on code in PR #59153:
URL: https://github.com/apache/spark/pull/59153#discussion_r4145110477
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -525,48 +531,83 @@ object SparkEnv extends Logging {
private[spark] val driverSystemName = "sparkDriver"
private[spark] val executorSystemName = "sparkExecutor"
+ /**
+ * Spark-owned implementation of direct worker creation. A string, not a
compile-time
+ * reference: `core` depends on `udf-worker-proto` and `udf-worker-core`
only, so the
+ * gRPC runtime stays out of this module. This is not a user extension point.
+ */
+ private[spark] val DIRECT_DISPATCHER_FACTORY_CLASS =
+ "org.apache.spark.udf.worker.grpc.DirectDispatcherFactory"
+
/**
* :: Experimental ::
- * Resolves the [[UDFDispatcherFactory]] used to create dispatchers for
external UDF
- * workers (SPARK-55278), from `spark.udf.worker.dispatcherFactory`.
- *
- * The factory is loaded reflectively rather than named here: `core` depends
on
- * `udf-worker-proto` and `udf-worker-core` only, so that neither it nor its
consumers
- * carry a gRPC dependency. Every concrete dispatcher therefore lives in a
module that
- * core cannot reference at compile time, and the implementation has to be
selected by
- * configuration.
+ * Resolves the Spark-owned [[UDFDispatcherFactory]] for external UDF workers
+ * (SPARK-55278). The factory routes from the closed `worker` oneof in
+ * [[UDFWorkerSpecification]], rather than from a second configuration value.
*
- * When the config is unset the returned factory throws on first use rather
than at
- * SparkEnv creation, so applications that never run an external UDF are
unaffected.
+ * A user-supplied class name is never loaded.
*/
- private[spark] def resolveUDFDispatcherFactory(
- conf: SparkConf,
- isDriver: Boolean): UDFDispatcherFactory = {
- conf.get(UDF.DISPATCHER_FACTORY) match {
- case Some(className) =>
- // Validate before constructing so that a misconfigured class is
reported against
- // the config key, rather than as a ClassCastException from the call
site.
- val cls = Utils.classForName[AnyRef](className, initialize = false)
- if (!classOf[UDFDispatcherFactory].isAssignableFrom(cls)) {
- throw new SparkException(
- s"${UDF.DISPATCHER_FACTORY.key} is set to $className, which does
not implement " +
- s"${classOf[UDFDispatcherFactory].getName}.")
- }
- Utils.instantiateSerializerOrShuffleManager[UDFDispatcherFactory](
- className, conf, isDriver)
-
- case None =>
- new UDFDispatcherFactory {
- override def createDispatcher(
- workerSpec: UDFWorkerSpecification,
- logger: WorkerLogger): WorkerDispatcher = {
+ private[spark] def resolveUDFDispatcherFactory(): UDFDispatcherFactory = {
+ new UDFDispatcherFactory {
+ override def createDispatcher(
+ workerSpec: UDFWorkerSpecification,
+ logger: WorkerLogger): WorkerDispatcher = {
+ workerSpec.getWorkerCase match {
+ case UDFWorkerSpecification.WorkerCase.DIRECT =>
+ loadBuiltInDispatcherFactory(
+ workerSpec.getWorkerCase.toString,
+ DIRECT_DISPATCHER_FACTORY_CLASS).createDispatcher(workerSpec,
logger)
+ case workerCase =>
throw new UnsupportedOperationException(
- "No UDF dispatcher factory is configured, so external UDF
workers cannot be " +
- s"created. Set ${UDF.DISPATCHER_FACTORY.key} to the name of a
" +
- s"${classOf[UDFDispatcherFactory].getName} implementation.")
- }
+ s"No built-in UDF dispatcher is available for worker type
$workerCase. " +
+ "Worker creation must be selected by
UDFWorkerSpecification.worker.")
}
+ }
+ }
+ }
+
+ /**
+ * Loads a Spark-owned factory. A missing runtime is reported against the
worker
+ * type, because `spark-udf-worker-grpc` is intentionally absent from the
engine
+ * classpath until its gRPC dependencies are shaded.
+ */
+ private def loadBuiltInDispatcherFactory(
+ workerType: String,
+ className: String): UDFDispatcherFactory = {
+ val cls = try {
+ Utils.classForName[AnyRef](className, initialize = false)
Review Comment:
I have one last concern about this reflection here, then i'm good with the
changes:
Could we load this through
classOf[SparkEnv].getClassLoader.loadClass(className) instead?
Utils.classForName uses the task context classloader, which includes user
artifacts. Since the gRPC module is intentionally absent, a user jar could
provide this exact class name—and with child-first classpaths could later
shadow Spark’s packaged implementation—defeating the Spark-owned-only guarantee.
Could we also cover this in SparkEnvUDFDispatcherSuite by running the
existing direct-worker test under Utils.withContextClassLoader with a loader
that throws if asked for DIRECT_DISPATCHER_FACTORY_CLASS? The fixed code should
bypass that loader and still produce the expected missing-runtime error.
--
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]
