[SPARK-56922] Direct dispatcher to manage grpc-based udf server.

[haiyangsun-db]

What changes were proposed in this pull request?

This PR builds on GrpcWorkerSession from #56702 and adds the transport and dispatcher components required to launch UDF workers and communicate with them over gRPC bidirectional streaming through Unix domain sockets.

The main changes are:

• Add DirectGrpcDispatcher, GrpcWorkerChannel, and native epoll/kqueue transport selection.
• Verify worker readiness using gRPC connectivity instead of socket-file existence.
• Centralize private socket-directory creation, path-length validation, and cleanup.
• Keep each direct worker process and its transport connection under one ownership boundary.
• Coordinate dispatcher shutdown with in-flight session creation and worker release so transport cleanup cannot race either operation.
• Use named daemon threads for gRPC event loops and terminate them during channel shutdown.
• Keep generic dispatcher lifecycle coverage in udf-worker-core and transport-specific integration coverage in udf-worker-grpc.
• Retain lightweight extension seams in EchoProtocolSuite for testing additional worker implementations.

Why are the changes needed?

GrpcWorkerSession implements an individual UDF execution stream, but Spark also needs a concrete dispatcher and transport to launch workers, establish connections, manage their lifecycle, and clean up resources.

Checking actual gRPC connectivity is important because a socket file may exist before the worker is ready to accept requests. Coordinating shutdown with both session creation and worker release prevents sessions from being published after close begins and prevents per-worker cleanup from racing shared transport cleanup.

Does this PR introduce any user-facing change?

No. These APIs are experimental and are not yet connected to a production Spark execution path.

How was this patch tested?

The worker module and SQL integration tests passed locally:

build/sbt "udf-worker-core/test" "udf-worker-grpc/test"
PYSPARK_PYTHON=python3.11 PYSPARK_DRIVER_PYTHON=python3.11 build/sbt "sql/testOnly org.apache.spark.sql.execution.externalUDF.PythonUDFWorkerSpecificationSuite"

After the lifecycle follow-up, the affected suites were rerun on Java 17:

build/sbt "udf-worker-core/testOnly *DirectWorkerDispatcherSuite" "udf-worker-grpc/testOnly *DirectGrpcDispatcherSuite" "udf-worker-grpc/testOnly *DirectGrpcDispatcherIntegrationSuite"

All 42 targeted tests passed. Main and test Scalastyle checks for both affected modules also passed.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

[HyukjinKwon]

Merge Summary:

• merged into master ad3b3b0
• merged into branch-4.x 66eefa0

Posted by merge_spark_pr.py