haiyangsun-db commented on code in PR #58859:
URL: https://github.com/apache/spark/pull/58859#discussion_r4122593130
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -225,17 +226,8 @@ class SparkEnv (
@volatile private var udfDispatcherManager: Option[UDFDispatcherManager] =
None
private def createUDFDispatcherManager(): UDFDispatcherManager = {
- val factory = new UDFDispatcherFactory {
- override def createDispatcher(
- workerSpec: UDFWorkerSpecification,
- logger: org.apache.spark.udf.worker.core.WorkerLogger
- ): WorkerDispatcher = {
- // TODO [SPARK-55278]: Wire in the correct dispatcher factory
- throw new UnsupportedOperationException(
- "No UDF dispatcher factory configured. " +
- "Set up a concrete factory for SPARK-55278.")
- }
- }
+ val factory = SparkEnv.resolveUDFDispatcherFactory(
Review Comment:
Coordinate lazy manager creation with `SparkEnv.stop()`. Running tasks
survive `threadPool.shutdown()`, so stop can miss a manager published afterward
and leave its workers unclosed. Use one lifecycle gate, reject creation after
`isStopped`, and cover this interleaving.
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -527,6 +519,50 @@ object SparkEnv extends Logging {
private[spark] val driverSystemName = "sparkDriver"
private[spark] val executorSystemName = "sparkExecutor"
+ /**
+ * :: 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.
+ *
+ * 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.
+ */
+ 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)
Review Comment:
`classForName` initializes by default, so an incompatible class can run or
fail before this check and bypass the targeted config error.
```suggestion
val cls = Utils.classForName[AnyRef](className, initialize = false)
```
--
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]
