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 <[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]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
