zozo123 commented on code in PR #59153:
URL: https://github.com/apache/spark/pull/59153#discussion_r4145415064
##########
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:
Good catch. Fixed in 336f81ce. The factory now loads through
`classOf[SparkEnv].getClassLoader`, so the task context loader is never
consulted. The DIRECT test in `SparkEnvUDFDispatcherSuite` now runs under a
context loader that throws if asked for `DIRECT_DISPATCHER_FACTORY_CLASS`. It
still gets the missing-runtime error. With the old `Utils.classForName` line,
the test fails with "User class loader was asked for
...DirectDispatcherFactory".
--
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]
