You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/06/18 18:12:53 UTC

(spark) branch branch-4.2 updated: [SPARK-57336][CONNECT] Send the bearer token in the standard Authorization header

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

dongjoon-hyun pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.2 by this push:
     new 2cfb0ab90688 [SPARK-57336][CONNECT] Send the bearer token in the standard Authorization header
2cfb0ab90688 is described below

commit 2cfb0ab9068846a88b6e20940eabdebfb3aac959
Author: Jiwon Park <jp...@outlook.kr>
AuthorDate: Thu Jun 18 11:04:29 2026 -0700

    [SPARK-57336][CONNECT] Send the bearer token in the standard Authorization header
    
    The Scala/JVM Spark Connect client sends the access token (from the `token` connection param) in a non-standard `Authentication` gRPC metadata header. This changes it to the standard `Authorization`, and adds a unit test.
    
    ```scala
    - Metadata.Key.of("Authentication", Metadata.ASCII_STRING_MARSHALLER)
    + Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER)
    ```
    
    Bearer tokens belong in `Authorization` (RFC 6750). The old key breaks token auth and is inconsistent with every other consumer:
    - server-side `PreSharedKeyAuthenticationInterceptor` reads the token from `Authorization`;
    
    https://github.com/apache/spark/blob/9d0b440554fe90d1b58c712c9ad1fe3c10bde6ec/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/PreSharedKeyAuthenticationInterceptor.scala#L28
    
    - the local-channel path in the same client already uses `Authorization`;
    - the Python client uses gRPC's `access_token_call_credentials` (i.e. `Authorization`);
    - standard JWT proxies (e.g. Envoy/Istio) extract the bearer from `Authorization` and otherwise reject the request.
    
    Yes. The Scala/JVM client now sends the token in `Authorization` instead of `Authentication`, fixing token auth against the pre-shared-key interceptor and standard JWT proxies.
    
    New unit test in `SparkConnectClientSuite` asserting `AccessTokenCallCredentials` emits `Authorization: Bearer <token>` (RED before the fix, GREEN after); full suite passes.
    
    Generated-by: Claude Code (Opus 4.8)
    
    Closes #56389 from j1wonpark/SPARK-57336.
    
    Authored-by: Jiwon Park <jp...@outlook.kr>
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
    (cherry picked from commit 4351d4626a090dad1bb996bc2927353c494a6e89)
    Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
 .../connect/client/SparkConnectClientSuite.scala   | 26 ++++++++++++++++++++--
 .../sql/connect/client/SparkConnectClient.scala    |  2 +-
 2 files changed, 25 insertions(+), 3 deletions(-)

diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala
index 22feaff1c77f..ceb561f69596 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/SparkConnectClientSuite.scala
@@ -18,13 +18,13 @@ package org.apache.spark.sql.connect.client
 
 import java.nio.charset.StandardCharsets.UTF_8
 import java.util.{Base64, UUID}
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{Executor, TimeUnit}
 
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
 
 import com.google.protobuf.{Any => PAny, StringValue}
-import io.grpc.{CallOptions, Channel, ClientCall, ClientInterceptor, Metadata, MethodDescriptor, Server, ServerCall, ServerCallHandler, ServerInterceptor, Status, StatusRuntimeException}
+import io.grpc.{CallCredentials, CallOptions, Channel, ClientCall, ClientInterceptor, Metadata, MethodDescriptor, Server, ServerCall, ServerCallHandler, ServerInterceptor, Status, StatusRuntimeException}
 import io.grpc.netty.NettyServerBuilder
 import io.grpc.stub.StreamObserver
 import org.scalatest.concurrent.Eventually
@@ -824,6 +824,28 @@ class SparkConnectClientSuite extends ConnectFunSuite {
       assert(!headerInterceptor.headers.exists(_.containsKey(key)))
     }
   }
+
+  test("SPARK-57336: access token is sent in the standard Authorization header") {
+    val token = "test-token-12345"
+    val creds = new SparkConnectClient.AccessTokenCallCredentials(token)
+
+    var captured: Option[Metadata] = None
+    var failure: Option[Status] = None
+    val applier = new CallCredentials.MetadataApplier {
+      override def apply(headers: Metadata): Unit = captured = Some(headers)
+      override def fail(status: Status): Unit = failure = Some(status)
+    }
+    val sameThreadExecutor = new Executor {
+      override def execute(command: Runnable): Unit = command.run()
+    }
+
+    creds.applyRequestMetadata(null, sameThreadExecutor, applier)
+
+    val authorizationKey = Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER)
+    assert(failure.isEmpty, s"unexpected failure: ${failure.orNull}")
+    assert(captured.isDefined)
+    assert(captured.get.get(authorizationKey) === s"Bearer $token")
+  }
 }
 
 class DummySparkConnectService() extends SparkConnectServiceGrpc.SparkConnectServiceImplBase {
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala
index d9b9ba35b5e6..de5e5e7744f2 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala
@@ -705,7 +705,7 @@ object SparkConnectClient {
   private val DEFAULT_USER_AGENT: String = "_SPARK_CONNECT_SCALA"
 
   private val AUTH_TOKEN_META_DATA_KEY: Metadata.Key[String] =
-    Metadata.Key.of("Authentication", Metadata.ASCII_STRING_MARSHALLER)
+    Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER)
 
   // for internal tests
   private[sql] def apply(channel: ManagedChannel): SparkConnectClient = {


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