[SPARK-59349][SQL] Override the gRPC authority for UDF worker Unix socket channels

[viirya]

What changes were proposed in this pull request?

GrpcWorkerChannel builds its gRPC channel over a Unix domain socket without calling overrideAuthority, so gRPC derives the HTTP/2 :authority pseudo-header from the target address. For a UDS that address is the socket path, which is not a valid HTTP authority.

This PR sets an explicit placeholder authority on the channel builder:

NettyChannelBuilder
  .forAddress(new DomainSocketAddress(socketPath))
  ...
  .overrideAuthority(GrpcWorkerChannel.UDS_AUTHORITY)   // added
  .usePlaintext()
  .build()

with private[grpc] val UDS_AUTHORITY: String = "localhost" in the companion object, plus a comment recording why, so the reasoning does not have to be rediscovered.

A placeholder is the correct value rather than a workaround: for a Unix socket peer the authority carries no information, because the socket path already identifies the peer completely. There is no name to resolve and no virtual host to select.

Why are the changes needed?

A conforming HTTP/2 server validates the pseudo-headers while decoding the HEADERS frame and resets the stream:

h2::server: malformed headers: malformed authority (b"var%2Ffolders%2F...%2Frw.sock"): invalid authority
h2::proto::streams::send: send_reset(..., reason=PROTOCOL_ERROR, initiator=Library, ...)

The client sees only an opaque protocol error:

io.grpc.StatusRuntimeException: INTERNAL: RST_STREAM closed stream. HTTP/2 error code: PROTOCOL_ERROR

The rejection happens before any worker application code runs -- not the service method, not an interceptor -- so the worker logs nothing and Init is never observed. The failure points at the wrong side of the connection while the worker has no way to report why, which makes it expensive to diagnose.

This is not reachable through any in-tree code path today. grpc-java's own server tolerates the malformed authority, and both Spark's client and the only in-tree worker (EchoGrpcWorkerMain) are grpc-java, so the pair works and the header is never validated. That is why it merged unnoticed in SPARK-56922, and it is also why it is worth fixing before the first third-party worker exists rather than after.

It does affect any worker on a conforming HTTP/2 stack, which is most stacks outside the JVM. worker_spec.proto invites workers in any language, and the SPARK-55278 SPIP names onboarding a Go/Rust/Swift worker as its final exam; such a worker hits this on its very first RPC.

This was confirmed three independent ways against a worker running on a conforming HTTP/2 server:

1. The same client succeeds against the same worker when only the authority changes -- grpc.insecure_channel(f"unix:{sock}", options=[("grpc.default_authority", "localhost")]) works where the default fails.
2. Driving that worker through Spark's own DirectGrpcDispatcher and GrpcWorkerSession reproduces the identical PROTOCOL_ERROR, so it is not a grpc-python quirk.
3. The server-side h2 trace above names the rejected header, so the cause is read from the library that rejects it rather than inferred.

Note the fix cannot go on the worker side. A server-side middleware layer that rewrites the request URI has no effect, because such layers run after header decode, by which point the stream is already reset. Fixing it below the HTTP/2 layer would mean rewriting the HPACK-encoded HEADERS frame, which is unreasonable for something the client sets with one call.

Does this PR introduce any user-facing change?

No. The affected code is unreleased (GrpcWorkerChannel arrived in SPARK-56922, fixVersion 4.4.0), and there is no behavior change for any in-tree code path -- grpc-java servers accept both the old and the new authority.

How was this patch tested?

New test in DirectGrpcDispatcherIntegrationSuite, which already spawns a real gRPC worker over a UDS, asserting the channel's configured authority:

val authority = channel.channel.authority()
assert(authority === GrpcWorkerChannel.UDS_AUTHORITY,
  s"expected the overridden authority, got '$authority'")

The test asserts on the configured authority rather than on an end-to-end round trip deliberately: an end-to-end test cannot detect this bug. The only in-tree worker is grpc-java, which tolerates the malformed authority, so a test driven through it passes with or without the fix. Asserting the authority tests the fix rather than the symptom.

I verified the test actually detects the defect by reverting the one-line fix and re-running it:

- the channel overrides the authority derived from the socket path *** FAILED ***
  "[/tmp/spark-udf-worker.../w-e59da31d51514d13.sock]" did not equal "[localhost]"
  expected the overridden authority, got '/tmp/spark-udf-worker.../w-e59da31d51514d13.sock'

It fails without the fix and passes with it.

End-to-end coverage against a non-grpc-java worker was verified separately: with the fix, a worker on a conforming HTTP/2 server driven through DirectGrpcDispatcher completes the protocol exchange that previously failed with PROTOCOL_ERROR.

Also run:

• build/sbt udf-worker-grpc/compile udf-worker-grpc/Test/compile
• build/sbt udf-worker-grpc/test -- 60 tests, all passing (4 suites, no pre-existing test disturbed)
• build/sbt udf-worker-grpc/scalastyle udf-worker-grpc/Test/scalastyle -- 0 errors

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

Generated-by: Claude Code

[viirya]

cc @haiyangsun-db

[viirya]

Merge Summary:

• merged into master 2cd436a
• merged into branch-4.x 98ae626

Posted by merge_spark_pr.py

[viirya]

Thanks @HyukjinKwon

[haiyangsun-db]

thank you for the fix!