zozo123 commented on code in PR #59153:
URL: https://github.com/apache/spark/pull/59153#discussion_r4143360631
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -525,50 +531,94 @@ object SparkEnv extends Logging {
private[spark] val driverSystemName = "sparkDriver"
private[spark] val executorSystemName = "sparkExecutor"
+ private[spark] val REMOVED_DISPATCHER_FACTORY_KEY =
"spark.udf.worker.dispatcherFactory"
+
+ /**
+ * 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. The old class-name setting was
+ * introduced after the last release and is rejected explicitly if present.
*/
- 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(conf: SparkConf):
UDFDispatcherFactory = {
+ if (conf.contains(REMOVED_DISPATCHER_FACTORY_KEY)) {
Review Comment:
Agreed. Removed the compatibility check. The setting never shipped, so this
is a forward fix.
--
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]
