This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new d69ef27ba5e1 [SPARK-59886][UDF] Route external UDF dispatch through
worker specifications
d69ef27ba5e1 is described below
commit d69ef27ba5e15dcfd2c70555e8aac0934f44d339
Author: Yossi Eliaz <[email protected]>
AuthorDate: Thu Oct 1 09:41:30 2026 +0900
[SPARK-59886][UDF] Route external UDF dispatch through worker specifications
### What changes were proposed in this pull request?
Replace the arbitrary `spark.udf.worker.dispatcherFactory` class hook
introduced by SPARK-59822 with Spark-owned routing from
`UDFWorkerSpecification.worker`.
The protobuf `oneof worker` already generates the closed `WorkerCase` enum,
so it is the single source of truth for worker creation mode. `DIRECT` maps to
a hard-coded Spark implementation, `DirectDispatcherFactory`, in
`spark-udf-worker-grpc`. `core` loads only that Spark-owned class name
reflectively because it intentionally has no compile-time dependency on the
gRPC module. User-provided class names are never loaded.
The unreleased class-name setting is dropped with no compatibility check.
The lazy dispatcher-manager lifecycle coordination from SPARK-59822 is
preserved.
### Why are the changes needed?
An arbitrary factory class executes operator-provided JVM code inside Spark
and makes engine behavior and security policy unpredictable. Worker-specific
commands, environment setup, capabilities, and connection requirements already
belong in `UDFWorkerSpecification` and execute outside the Spark JVM.
A separate dispatcher enum config would duplicate `WorkerCase` and could
disagree with the specification. Routing from the existing closed `oneof` keeps
Spark in control of dispatch while leaving worker customization in the
worker-spec contract.
### Does this PR introduce _any_ user-facing change?
Yes, relative to unreleased master and branch-4.x.
`spark.udf.worker.dispatcherFactory` is removed; custom dispatcher classes are
no longer supported. A `DirectWorker` specification selects Spark's built-in
direct dispatcher automatically.
The gRPC implementation is still outside Spark's runtime classpath until
SPARK-57214 packages and shades it, so direct execution currently reports an
actionable missing-runtime error. This PR does not make gRPC a dependency of
`core`.
### Follow-up direction
No further Spark API is needed for the first sandbox proof. `DirectWorker`
already names an external local process and assigns it a Unix domain socket.
That process can launch the standard UDF worker inside a local Docker sandbox,
with dependencies present only in the sandbox. Sandbox-specific APIs do not
belong in Spark, and this PR does not add them.
### How was this patch tested?
- `core/testOnly org.apache.spark.SparkEnvUDFDispatcherSuite`
- `udf-worker-grpc/testOnly
org.apache.spark.udf.worker.grpc.DirectGrpcDispatcherSuite`
- `sql/testOnly
org.apache.spark.sql.execution.externalUDF.PlanExternalUDFsSuite`
The suites cover unset worker types, `DIRECT` routing, shutdown racing with
lazy manager creation, concrete factory construction, and the SQL execution
boundary.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Auto
Closes #59153 from zozo123/SPARK-59886.
Authored-by: Yossi Eliaz <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 56da062b59f8ae58faa81554c76de49937d2d232)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../src/main/scala/org/apache/spark/SparkEnv.scala | 117 ++++++++++++-------
.../org/apache/spark/internal/config/UDF.scala | 34 ------
.../apache/spark/SparkEnvUDFDispatcherSuite.scala | 125 +++++++--------------
.../externalUDF/PlanExternalUDFsSuite.scala | 5 +-
udf/worker/README.md | 6 +
.../udf/worker/grpc/DirectDispatcherFactory.scala | 39 +++++++
.../worker/grpc/DirectGrpcDispatcherSuite.scala | 15 +++
7 files changed, 181 insertions(+), 160 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala
b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index 04b75c4bcee3..d6455ac3b4c5 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -226,10 +226,16 @@ class SparkEnv (
// Guarded by this SparkEnv's monitor, together with isStopped.
private var udfDispatcherManager: Option[UDFDispatcherManager] = None
+ /**
+ * Builds the factory for this env. Overridable from tests so the shutdown
race
+ * can be exercised without a user-supplied class name.
+ */
+ private[spark] def createUDFDispatcherFactory(): UDFDispatcherFactory = {
+ SparkEnv.resolveUDFDispatcherFactory()
+ }
+
private def createUDFDispatcherManager(): UDFDispatcherManager = {
- val factory = SparkEnv.resolveUDFDispatcherFactory(
- conf, SparkContext.isDriver(executorId))
- new UDFDispatcherManager(factory, new SparkUDFWorkerLogger())
+ new UDFDispatcherManager(createUDFDispatcherFactory(), new
SparkUDFWorkerLogger())
}
/**
@@ -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 {
+ classOf[SparkEnv].getClassLoader.loadClass(className)
+ } catch {
+ case _: ClassNotFoundException =>
+ throw dispatcherRuntimeUnavailable(workerType, className)
+ case _: LinkageError =>
+ throw dispatcherRuntimeUnavailable(workerType, className)
}
+ if (!classOf[UDFDispatcherFactory].isAssignableFrom(cls)) {
+ throw new SparkException(
+ s"Worker type $workerType resolved to $className, which does not
implement " +
+ s"${classOf[UDFDispatcherFactory].getName}.")
+ }
+ try {
+ cls.getConstructor().newInstance().asInstanceOf[UDFDispatcherFactory]
+ } catch {
+ case _: ClassNotFoundException =>
+ throw dispatcherRuntimeUnavailable(workerType, className)
+ case _: LinkageError =>
+ throw dispatcherRuntimeUnavailable(workerType, className)
+ case e: Exception =>
+ throw new SparkException(
+ s"Could not construct the built-in dispatcher factory $className for
worker " +
+ s"type $workerType.", e)
+ }
+ }
+
+ private def dispatcherRuntimeUnavailable(
+ workerType: String,
+ className: String): SparkException = {
+ new SparkException(
+ s"Spark could not load the built-in dispatcher implementation
($className) for " +
+ s"worker type $workerType. The direct dispatcher comes from the " +
+ "spark-udf-worker-grpc module, which is not on the Spark runtime
classpath yet.")
}
def set(e: SparkEnv): Unit = {
diff --git a/core/src/main/scala/org/apache/spark/internal/config/UDF.scala
b/core/src/main/scala/org/apache/spark/internal/config/UDF.scala
deleted file mode 100644
index af536cd5fb0b..000000000000
--- a/core/src/main/scala/org/apache/spark/internal/config/UDF.scala
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.internal.config
-
-private[spark] object UDF {
-
- val DISPATCHER_FACTORY = ConfigBuilder("spark.udf.worker.dispatcherFactory")
- .doc("Name of the class implementing " +
- "org.apache.spark.udf.worker.core.UDFDispatcherFactory, used to create
the " +
- "dispatchers that provision and manage external UDF workers
(SPARK-55278). The " +
- "class must have a no-arg constructor, a constructor taking a SparkConf,
or a " +
- "constructor taking a SparkConf and a Boolean indicating whether it is
being " +
- "created on the driver. When unset, no external UDF dispatcher is
available and " +
- "executing an external UDF fails. This is an operator-level setting: it
selects " +
- "the runtime that user code is executed by, so it is not intended to be
changed " +
- "by individual queries.")
- .version("4.4.0")
- .stringConf
- .createOptional
-}
diff --git
a/core/src/test/scala/org/apache/spark/SparkEnvUDFDispatcherSuite.scala
b/core/src/test/scala/org/apache/spark/SparkEnvUDFDispatcherSuite.scala
index 886bbafbbdc8..43cdbe8edfea 100644
--- a/core/src/test/scala/org/apache/spark/SparkEnvUDFDispatcherSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkEnvUDFDispatcherSuite.scala
@@ -26,20 +26,21 @@ import org.mockito.Answers.RETURNS_DEEP_STUBS
import org.mockito.Mockito.{mock, verify}
import org.apache.spark.broadcast.BroadcastManager
-import org.apache.spark.internal.config.UDF
import org.apache.spark.metrics.MetricsSystem
import org.apache.spark.rpc.RpcEnv
import org.apache.spark.scheduler.OutputCommitCoordinator
import org.apache.spark.storage.BlockManager
-import org.apache.spark.udf.worker.UDFWorkerSpecification
+import org.apache.spark.udf.worker.{DirectWorker, UDFWorkerSpecification}
import org.apache.spark.udf.worker.core.{UDFDispatcherFactory,
WorkerDispatcher, WorkerLogger}
-import org.apache.spark.util.ThreadUtils
+import org.apache.spark.util.{ThreadUtils, Utils}
class SparkEnvUDFDispatcherSuite extends SparkFunSuite {
private def spec = UDFWorkerSpecification.getDefaultInstance
- private def newEnv(conf: SparkConf): SparkEnv = {
+ private def newEnv(
+ conf: SparkConf,
+ factory: Option[() => UDFDispatcherFactory] = None): SparkEnv = {
new SparkEnv(
SparkContext.DRIVER_IDENTIFIER,
mock(classOf[RpcEnv]),
@@ -52,74 +53,54 @@ class SparkEnvUDFDispatcherSuite extends SparkFunSuite {
null,
mock(classOf[MetricsSystem]),
mock(classOf[OutputCommitCoordinator]),
- conf)
+ conf) {
+ override private[spark] def createUDFDispatcherFactory():
UDFDispatcherFactory = {
+ factory.map(_()).getOrElse(super.createUDFDispatcherFactory())
+ }
+ }
}
- test("no factory configured: dispatcher creation fails with an actionable
message") {
- val factory = SparkEnv.resolveUDFDispatcherFactory(new SparkConf(false),
isDriver = true)
- // Resolution itself must succeed: applications that never run an external
UDF should not
- // fail at SparkEnv creation just because no dispatcher is configured.
+ test("an unset worker type fails with an actionable message") {
+ val factory = SparkEnv.resolveUDFDispatcherFactory()
val e = intercept[UnsupportedOperationException] {
factory.createDispatcher(spec, WorkerLogger.NoOp)
}
- assert(e.getMessage.contains(UDF.DISPATCHER_FACTORY.key))
- }
-
- test("configured factory is instantiated and used") {
- val conf = new SparkConf(false)
- .set(UDF.DISPATCHER_FACTORY, classOf[TestNoArgDispatcherFactory].getName)
- val factory = SparkEnv.resolveUDFDispatcherFactory(conf, isDriver = true)
- assert(factory.isInstanceOf[TestNoArgDispatcherFactory])
- assert(factory.createDispatcher(spec, WorkerLogger.NoOp) === null)
- }
-
- test("factory receives the SparkConf and isDriver when it declares them") {
- val conf = new SparkConf(false)
- .set(UDF.DISPATCHER_FACTORY, classOf[TestConfDispatcherFactory].getName)
- .set("spark.test.marker", "set-by-conf")
- Seq(true, false).foreach { isDriver =>
- val factory = SparkEnv.resolveUDFDispatcherFactory(conf, isDriver)
- .asInstanceOf[TestConfDispatcherFactory]
- assert(factory.conf.get("spark.test.marker") === "set-by-conf")
- assert(factory.isDriver === isDriver)
- }
+ assert(e.getMessage.contains("WORKER_NOT_SET"))
+ assert(e.getMessage.contains("UDFWorkerSpecification.worker"))
}
- test("a class that is not a UDFDispatcherFactory is rejected against the
config key") {
- val conf = new SparkConf(false)
- .set(UDF.DISPATCHER_FACTORY, classOf[NotAFactory].getName)
- val e = intercept[SparkException] {
- SparkEnv.resolveUDFDispatcherFactory(conf, isDriver = true)
+ test("a direct worker requests the Spark-owned runtime") {
+ val directSpec = UDFWorkerSpecification
+ .newBuilder()
+ .setDirect(DirectWorker.getDefaultInstance)
+ .build()
+ val userLoader = new ClassLoader(getClass.getClassLoader) {
+ override def loadClass(name: String, resolve: Boolean): Class[_] = {
+ if (name == SparkEnv.DIRECT_DISPATCHER_FACTORY_CLASS) {
+ throw new IllegalStateException(s"User class loader was asked for
$name")
+ }
+ super.loadClass(name, resolve)
+ }
}
- assert(e.getMessage.contains(UDF.DISPATCHER_FACTORY.key))
- assert(e.getMessage.contains(classOf[NotAFactory].getName))
- }
-
- test("an incompatible factory class is rejected before its initializer
runs") {
- val key = "spark.test.incompatibleFactoryInitialized"
- System.clearProperty(key)
- try {
- val conf = new SparkConf(false)
- .set(UDF.DISPATCHER_FACTORY,
"org.apache.spark.NotAFactoryWithInitializer$")
+ val factory = SparkEnv.resolveUDFDispatcherFactory()
+ val e = Utils.withContextClassLoader(userLoader) {
intercept[SparkException] {
- SparkEnv.resolveUDFDispatcherFactory(conf, isDriver = true)
+ factory.createDispatcher(directSpec, WorkerLogger.NoOp)
}
- assert(System.getProperty(key) == null)
- } finally {
- System.clearProperty(key)
}
+ assert(e.getMessage.contains("DIRECT"))
+ assert(e.getMessage.contains("spark-udf-worker-grpc"))
+ assert(e.getMessage.contains(SparkEnv.DIRECT_DISPATCHER_FACTORY_CLASS))
}
test("stop waits for lazy dispatcher manager creation and closes created
dispatchers") {
- val factory = TestBlockingDispatcherFactory
- val conf = new SparkConf(false)
- .set(UDF.DISPATCHER_FACTORY,
classOf[TestBlockingDispatcherFactory].getName)
- val env = newEnv(conf)
+ val hooks = TestBlockingDispatcherFactory
+ val env = newEnv(new SparkConf(false), Some(() => new
TestBlockingDispatcherFactory))
val pool = ThreadUtils.newDaemonFixedThreadPool(2, "udf-dispatcher-test")
implicit val executionContext: ExecutionContext =
ExecutionContext.fromExecutor(pool)
val creator = Future { Try(env.getExternalUDFDispatcher(spec)) }
try {
- assert(factory.entered.await(10, TimeUnit.SECONDS))
+ assert(hooks.entered.await(10, TimeUnit.SECONDS))
val stopStarted = new CountDownLatch(1)
val stopFinished = new CountDownLatch(1)
val stopper = Future {
@@ -129,49 +110,21 @@ class SparkEnvUDFDispatcherSuite extends SparkFunSuite {
}
assert(stopStarted.await(10, TimeUnit.SECONDS))
assert(!stopFinished.await(100, TimeUnit.MILLISECONDS))
- factory.release.countDown()
+ hooks.release.countDown()
val result = ThreadUtils.awaitResult(creator, 30.seconds)
ThreadUtils.awaitResult(stopper, 30.seconds)
assert(result.isSuccess ||
result.failed.get.isInstanceOf[IllegalStateException])
- if (factory.dispatcherCreated) {
- verify(factory.dispatcher).close()
+ if (hooks.dispatcherCreated) {
+ verify(hooks.dispatcher).close()
}
intercept[IllegalStateException] {
env.getExternalUDFDispatcher(spec)
}
} finally {
- factory.release.countDown()
+ hooks.release.countDown()
pool.shutdownNow()
}
}
-
- test("an unknown class name fails to resolve") {
- val conf = new SparkConf(false).set(UDF.DISPATCHER_FACTORY,
"not.a.real.Factory")
- intercept[ClassNotFoundException] {
- SparkEnv.resolveUDFDispatcherFactory(conf, isDriver = true)
- }
- }
-}
-
-// Declared as top-level classes so that they have constructors Spark can find
reflectively.
-
-class TestNoArgDispatcherFactory extends UDFDispatcherFactory {
- override def createDispatcher(
- workerSpec: UDFWorkerSpecification,
- logger: WorkerLogger): WorkerDispatcher = null
-}
-
-class TestConfDispatcherFactory(val conf: SparkConf, val isDriver: Boolean)
- extends UDFDispatcherFactory {
- override def createDispatcher(
- workerSpec: UDFWorkerSpecification,
- logger: WorkerLogger): WorkerDispatcher = null
-}
-
-class NotAFactory
-
-object NotAFactoryWithInitializer {
- System.setProperty("spark.test.incompatibleFactoryInitialized", "true")
}
object TestBlockingDispatcherFactory {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
index 39a70100f82a..f9ef2a8253d0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
@@ -732,8 +732,9 @@ class PlanExternalUDFsSuite extends QueryTest with
SharedSparkSession {
execution.execute().collect()
}
val message = Utils.exceptionString(exception)
- assert(message.contains("No UDF dispatcher factory is configured"))
- assert(message.contains("spark.udf.worker.dispatcherFactory"))
+ assert(message.contains("built-in dispatcher implementation"))
+ assert(message.contains("DIRECT"))
+ assert(message.contains("spark-udf-worker-grpc"))
}
test("map partitions external UDF execution reports unimplemented before
worker startup") {
diff --git a/udf/worker/README.md b/udf/worker/README.md
index 50ee7c6a60bc..9d76828ae32f 100644
--- a/udf/worker/README.md
+++ b/udf/worker/README.md
@@ -29,6 +29,12 @@ framework currently provides **direct worker creation**
(local OS
processes) and is designed for future **indirect creation** (via a
provisioning service or daemon).
+Spark selects its built-in implementation from the closed `worker`
+choice in `UDFWorkerSpecification`; `DirectWorker` selects local-process
+dispatch. The worker command, environment, capabilities, and connection
+belong in that specification. Spark does not load a user-supplied
+dispatcher class into the engine.
+
## Sub-packages
```
diff --git
a/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/DirectDispatcherFactory.scala
b/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/DirectDispatcherFactory.scala
new file mode 100644
index 000000000000..99cbe193647e
--- /dev/null
+++
b/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/DirectDispatcherFactory.scala
@@ -0,0 +1,39 @@
+/*
+ * 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.udf.worker.grpc
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.udf.worker.UDFWorkerSpecification
+import org.apache.spark.udf.worker.core.{UDFDispatcherFactory,
WorkerDispatcher, WorkerLogger}
+
+/**
+ * :: Experimental :: Spark-owned factory for the `DIRECT` dispatch mode.
+ *
+ * `core` cannot name this class at compile time, because that would put the
gRPC runtime on the
+ * engine classpath. [[org.apache.spark.SparkEnv]] loads this no-arg class for
a `DirectWorker`
+ * specification. It is not a user extension point: worker-specific behavior
belongs in
+ * [[UDFWorkerSpecification]].
+ */
+@Experimental
+private[spark] class DirectDispatcherFactory extends UDFDispatcherFactory {
+
+ override def createDispatcher(
+ workerSpec: UDFWorkerSpecification,
+ logger: WorkerLogger): WorkerDispatcher = {
+ new DirectGrpcDispatcher(workerSpec, logger)
+ }
+}
diff --git
a/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherSuite.scala
b/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherSuite.scala
index 70640c720d5a..319d045b775b 100644
---
a/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherSuite.scala
+++
b/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherSuite.scala
@@ -23,6 +23,7 @@ import org.scalatest.funsuite.AnyFunSuite
import org.apache.spark.udf.worker.{DirectWorker, LocalTcpConnection,
UDFProtoCommunicationPattern, UDFWorkerProperties, UDFWorkerSpecification,
UnixDomainSocket, WorkerCapabilities, WorkerConnectionSpec}
+import org.apache.spark.udf.worker.core.WorkerLogger
class DirectGrpcDispatcherSuite extends AnyFunSuite {
// scalastyle:on funsuite
@@ -79,6 +80,20 @@ class DirectGrpcDispatcherSuite extends AnyFunSuite {
assert(error.getMessage.contains("requires BIDIRECTIONAL_STREAMING"))
}
+ test("DIRECT factory creates a DirectGrpcDispatcher") {
+ val capabilities = WorkerCapabilities.newBuilder()
+ .addSupportedCommunicationPatterns(
+ UDFProtoCommunicationPattern.BIDIRECTIONAL_STREAMING)
+ .build()
+ val dispatcher = new DirectDispatcherFactory().createDispatcher(
+ directSpec(udsProperties, Some(capabilities)), WorkerLogger.NoOp)
+ try {
+ assert(dispatcher.isInstanceOf[DirectGrpcDispatcher])
+ } finally {
+ dispatcher.close()
+ }
+ }
+
test("requests a backoff reset for every transient failure after endpoint
creation") {
assert((1 to 2).forall { _ =>
DirectGrpcDispatcher.shouldResetConnectBackoff(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
