(spark) branch branch-4.x updated: [SPARK-59349][SQL] Override the gRPC authority for UDF worker Unix socket channels

[vi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

viirya 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 98ae626aba05 [SPARK-59349][SQL] Override the gRPC authority for UDF worker Unix socket channels
98ae626aba05 is described below

commit 98ae626aba05d072623fc433bb7e02b5614dfb55
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Tue Sep 8 22:46:33 2026 -0700

    [SPARK-59349][SQL] Override the gRPC authority for UDF worker Unix socket channels
    
    ### 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:
    
    ```scala
    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:
    
    ```scala
    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
    
    Closes #58638 from viirya/uds-authority.
    
    Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
    Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
    (cherry picked from commit 2cd436aada3572200837a31581cae1e4534aaad3)
    Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
 .../spark/udf/worker/grpc/GrpcWorkerChannel.scala      | 18 ++++++++++++++++++
 .../grpc/DirectGrpcDispatcherIntegrationSuite.scala    | 18 ++++++++++++++++++
 2 files changed, 36 insertions(+)

diff --git a/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/GrpcWorkerChannel.scala b/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/GrpcWorkerChannel.scala
index 220a31d63331..42686d6737eb 100644
--- a/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/GrpcWorkerChannel.scala
+++ b/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/GrpcWorkerChannel.scala
@@ -77,6 +77,7 @@ class GrpcWorkerChannel(
       .channelType(transport.channelType)
       .eventLoopGroup(eventLoopGroup)
       .maxInboundMessageSize(GrpcWorkerChannel.MAX_INBOUND_MESSAGE_SIZE)
+      .overrideAuthority(GrpcWorkerChannel.UDS_AUTHORITY)
       .usePlaintext()
       .build()
   } catch {
@@ -132,5 +133,22 @@ object GrpcWorkerChannel {
   /** Matches Spark Connect's default and accommodates normal Arrow batches. */
   private[grpc] val MAX_INBOUND_MESSAGE_SIZE: Int = 128 * 1024 * 1024
 
+  /**
+   * Authority sent in the HTTP/2 `:authority` pseudo-header.
+   *
+   * Without an override, gRPC derives the authority from the target address, which for a
+   * Unix domain socket is the socket path (percent-encoded where the path needs escaping,
+   * e.g. `var%2Ffolders%2F...%2Frw.sock`). Either form is an invalid HTTP authority. A
+   * conforming HTTP/2 server rejects such a HEADERS frame with PROTOCOL_ERROR while
+   * decoding it, so the RPC fails before any worker code -- service method or interceptor
+   * -- runs, and the worker cannot report why. grpc-java's own server accepts it, which is
+   * why the in-tree grpc-java test worker does not surface this.
+   *
+   * A placeholder is the right value rather than a workaround: the authority carries no
+   * information for a Unix socket peer, because the socket path already identifies it
+   * completely. There is no name to resolve and no virtual host to select.
+   */
+  private[grpc] val UDS_AUTHORITY: String = "localhost"
+
   private val SHUTDOWN_PHASE_TIMEOUT_MS: Long = 5000L
 }
diff --git a/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala b/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala
index 370bf907414c..7ede7a635904 100644
--- a/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala
+++ b/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/DirectGrpcDispatcherIntegrationSuite.scala
@@ -160,6 +160,24 @@ class DirectGrpcDispatcherIntegrationSuite
     assert(!socketDir.exists(), "dispatcher close should remove its socket directory")
   }
 
+  test("the channel overrides the authority derived from the socket path") {
+    dispatcher = new DirectGrpcDispatcher(workerSpec())
+    val session = dispatcher.createSession(None)
+    val channel = grpcChannel(workerProcess(session))
+
+    try {
+      // Without the override the authority would be the socket path, which a conforming
+      // HTTP/2 server rejects with PROTOCOL_ERROR while decoding the HEADERS frame.
+      // The in-tree worker is grpc-java, which tolerates the encoded form, so this asserts on
+      // the channel's configured authority rather than on an end-to-end failure.
+      val authority = channel.channel.authority()
+      assert(authority === GrpcWorkerChannel.UDS_AUTHORITY,
+        s"expected the overridden authority, got '$authority'")
+    } finally {
+      session.close(emptyCancel)
+    }
+  }
+
   test("a socket path alone does not make a worker ready") {
     val socketOnlyWorker =
       """


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org