This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 69b861c0d4c6 [SPARK-57336][CONNECT] Send the bearer token in the
standard Authorization header
69b861c0d4c6 is described below
commit 69b861c0d4c685d443fc347bb92bec4fe225b5a2
Author: Jiwon Park <[email protected]>
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 <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 4351d4626a090dad1bb996bc2927353c494a6e89)
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 2cfb0ab9068846a88b6e20940eabdebfb3aac959)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../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 743112c6dd4d..81ff034002be 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
@@ -17,12 +17,12 @@
package org.apache.spark.sql.connect.client
import java.util.UUID
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{Executor, TimeUnit}
import scala.collection.mutable
import scala.jdk.CollectionConverters._
-import io.grpc.{CallOptions, Channel, ClientCall, ClientInterceptor,
MethodDescriptor, Server, Status, StatusRuntimeException}
+import io.grpc.{CallCredentials, CallOptions, Channel, ClientCall,
ClientInterceptor, Metadata, MethodDescriptor, Server, Status,
StatusRuntimeException}
import io.grpc.netty.NettyServerBuilder
import io.grpc.stub.StreamObserver
import org.scalatest.BeforeAndAfterEach
@@ -620,6 +620,28 @@ class SparkConnectClientSuite extends ConnectFunSuite with
BeforeAndAfterEach {
// The client should try to fetch the config only once.
assert(service.getAndClearLatestConfigRequests().size == 1)
}
+
+ 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 5d36fc45f948..f6a413bd76b9 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
@@ -667,7 +667,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: [email protected]
For additional commands, e-mail: [email protected]
