This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 c167c6687f94 [SPARK-58632][CONNECT] Send Spark Connect operation ID in
gRPC metadata
c167c6687f94 is described below
commit c167c6687f94fcb4ef7595514df8ad436972c807
Author: Wenchen Fan <[email protected]>
AuthorDate: Thu Aug 20 14:24:10 2026 +0800
[SPARK-58632][CONNECT] Send Spark Connect operation ID in gRPC metadata
### What changes were proposed in this pull request?
Spark Connect's Scala and Python clients now send the `ExecutePlan`
operation ID in the
`spark-connect-operation-id` gRPC metadata header. The same operation ID is
used for initial,
reattach, and release RPCs. Client connection options cannot override this
per-operation value.
### Why are the changes needed?
Gateways and other intermediaries cannot reliably correlate an
`ExecutePlan` request by inspecting
the protobuf body, especially when a request fails before the server
returns a response. A stable,
vendor-neutral metadata header makes the client-generated operation ID
available throughout the
request path without decoding the payload.
### Does this PR introduce _any_ user-facing change?
Yes. Spark Connect clients now include `spark-connect-operation-id` in gRPC
metadata for operation
RPCs. The value matches `ExecutePlanRequest.operation_id`.
### How was this patch tested?
Added Scala tests for reattachable and non-reattachable clients and a
Python test that verifies
connection metadata cannot override the operation ID.
Attempted to run:
`build/sbt 'connect-client-jvm/testOnly
org.apache.spark.sql.connect.client.SparkConnectClientSuite'`
The test could not start because Maven Central DNS resolution was
unavailable in the local
environment.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #57834 from cloud-fan/cloud-fan/spark-connect-operation-id-header.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit bdcb08abf260cb11fedf91e88b514e42351c1b63)
Signed-off-by: Wenchen Fan <[email protected]>
---
python/pyspark/sql/connect/client/core.py | 56 +++++++++----
python/pyspark/sql/connect/plan.py | 2 +-
.../sql/tests/connect/client/test_client.py | 62 ++++++++++++++
.../connect/client/SparkConnectClientSuite.scala | 96 ++++++++++++++++++----
.../client/CustomSparkConnectBlockingStub.scala | 14 +++-
.../ExecutePlanResponseReattachableIterator.scala | 10 ++-
.../sql/connect/client/SparkConnectClient.scala | 16 +++-
7 files changed, 221 insertions(+), 35 deletions(-)
diff --git a/python/pyspark/sql/connect/client/core.py
b/python/pyspark/sql/connect/client/core.py
index 462ad8fe2670..03bd25e9c34e 100644
--- a/python/pyspark/sql/connect/client/core.py
+++ b/python/pyspark/sql/connect/client/core.py
@@ -130,6 +130,7 @@ if TYPE_CHECKING:
PYSPARK_ROOT = os.path.dirname(pyspark.__file__)
+_OPERATION_ID_METADATA_KEY = "spark-connect-operation-id"
@dataclass(frozen=True)
@@ -887,6 +888,16 @@ class SparkConnectClient(object):
if isinstance(connection, ChannelBuilder)
else DefaultChannelBuilder(connection, channel_options)
)
+ metadata = list(self._builder.metadata())
+ if any(key.lower() == _OPERATION_ID_METADATA_KEY for key, _ in
metadata):
+ logger.warning(
+ "Connection option %s is ignored because Spark Connect sets it
for each "
+ "ExecutePlan request.",
+ _OPERATION_ID_METADATA_KEY,
+ )
+ artifact_manager_metadata = [
+ (key, value) for key, value in metadata if key.lower() !=
_OPERATION_ID_METADATA_KEY
+ ]
self._user_id = None
self._retry_policies: List[RetryPolicy] = []
@@ -927,7 +938,7 @@ class SparkConnectClient(object):
self._user_id,
self._session_id,
self._channel,
- self._builder.metadata(),
+ artifact_manager_metadata,
add_artifacts_timeout=self._rpc_deadlines.add_artifacts,
artifact_status_timeout=self._rpc_deadlines.artifact_status,
)
@@ -1657,7 +1668,7 @@ class SparkConnectClient(object):
with attempt:
resp = self._stub.AnalyzePlan(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.analyze_plan,
)
self._verify_response_integrity(resp)
@@ -1693,7 +1704,7 @@ class SparkConnectClient(object):
req,
self._stub,
self._retrying,
- self._builder.metadata(),
+ self._execute_plan_metadata(req.operation_id),
reattachable_execute_plan_timeout=self._rpc_deadlines.reattachable_execute_plan,
reattach_execute_timeout=self._rpc_deadlines.reattach_execute,
)
@@ -1707,11 +1718,25 @@ class SparkConnectClient(object):
for attempt in self._retrying():
with attempt:
with disable_gc():
- for b in self._stub.ExecutePlan(req,
metadata=self._builder.metadata()):
+ for b in self._stub.ExecutePlan(
+ req,
metadata=self._execute_plan_metadata(req.operation_id)
+ ):
handle_response(b)
except Exception as error:
self._handle_error(error, req.operation_id)
+ def _builder_metadata(self) -> List[Tuple[str, str]]:
+ return [
+ (key, value)
+ for key, value in self._builder.metadata()
+ if key.lower() != _OPERATION_ID_METADATA_KEY
+ ]
+
+ def _execute_plan_metadata(self, operation_id: str) -> List[Tuple[str,
str]]:
+ metadata = self._builder_metadata()
+ metadata.append((_OPERATION_ID_METADATA_KEY, operation_id))
+ return metadata
+
def _execute_and_fetch_as_iterator(
self,
req: pb2.ExecutePlanRequest,
@@ -1735,7 +1760,6 @@ class SparkConnectClient(object):
for hook in self._session_hooks:
req = hook.on_execute_plan(req)
req.operation_id = operation_id
-
num_records = 0
arrow_batch_chunks_to_assemble: List[bytes] = []
@@ -1910,7 +1934,7 @@ class SparkConnectClient(object):
req,
self._stub,
self._retrying,
- self._builder.metadata(),
+ self._execute_plan_metadata(req.operation_id),
reattachable_execute_plan_timeout=self._rpc_deadlines.reattachable_execute_plan,
reattach_execute_timeout=self._rpc_deadlines.reattach_execute,
)
@@ -1925,7 +1949,9 @@ class SparkConnectClient(object):
with attempt:
with disable_gc():
it = iter(
- self._stub.ExecutePlan(req,
metadata=self._builder.metadata())
+ self._stub.ExecutePlan(
+ req,
metadata=self._execute_plan_metadata(req.operation_id)
+ )
)
while True:
try:
@@ -2067,7 +2093,7 @@ class SparkConnectClient(object):
with disable_gc():
resp = self._stub.Config(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.config,
)
self._verify_response_integrity(resp)
@@ -2113,7 +2139,7 @@ class SparkConnectClient(object):
with attempt:
resp = self._stub.Interrupt(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.interrupt,
)
self._verify_response_integrity(resp)
@@ -2129,7 +2155,7 @@ class SparkConnectClient(object):
with attempt:
resp = self._stub.Interrupt(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.interrupt,
)
self._verify_response_integrity(resp)
@@ -2145,7 +2171,7 @@ class SparkConnectClient(object):
with attempt:
resp = self._stub.Interrupt(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.interrupt,
)
self._verify_response_integrity(resp)
@@ -2165,7 +2191,7 @@ class SparkConnectClient(object):
with attempt:
resp = self._stub.ReleaseSession(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.release_session,
)
self._verify_response_integrity(resp)
@@ -2221,7 +2247,7 @@ class SparkConnectClient(object):
with attempt:
resp = self._stub.GetStatus(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.get_status,
)
self._verify_response_integrity(resp)
@@ -2357,7 +2383,7 @@ class SparkConnectClient(object):
try:
return self._stub.FetchErrorDetails(
req,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.fetch_error_details,
)
except grpc.RpcError:
@@ -2776,7 +2802,7 @@ class SparkConnectClient(object):
with attempt:
response: pb2.CloneSessionResponse = self._stub.CloneSession(
request,
- metadata=self._builder.metadata(),
+ metadata=self._builder_metadata(),
timeout=self._rpc_deadlines.clone_session,
)
diff --git a/python/pyspark/sql/connect/plan.py
b/python/pyspark/sql/connect/plan.py
index 51f117516663..709071b8acc2 100644
--- a/python/pyspark/sql/connect/plan.py
+++ b/python/pyspark/sql/connect/plan.py
@@ -784,7 +784,7 @@ class CachedRemoteRelation(LogicalPlan):
request_serializer=request_serializer,
response_deserializer=response_deserializer,
)
- metadata = session.client._builder.metadata()
+ metadata =
session.client._execute_plan_metadata(req.operation_id)
channel(req, metadata=metadata) # type:
ignore[arg-type]
except Exception as e:
logger.warning(f"RemoveRemoteCachedRelation failed with
exception: {e}.")
diff --git a/python/pyspark/sql/tests/connect/client/test_client.py
b/python/pyspark/sql/tests/connect/client/test_client.py
index 35252ed799ea..5e24bd686517 100644
--- a/python/pyspark/sql/tests/connect/client/test_client.py
+++ b/python/pyspark/sql/tests/connect/client/test_client.py
@@ -113,16 +113,22 @@ if should_test_connect:
self.release_calls = 0
self.release_until_calls = 0
self.attach_calls = 0
+ self.execute_metadata = []
+ self.attach_metadata = []
+ self.release_metadata = []
def ExecutePlan(self, *args, **kwargs):
self.execute_calls += 1
+ self.execute_metadata.append(kwargs.get("metadata"))
return self._execute_ops
def ReattachExecute(self, *args, **kwargs):
self.attach_calls += 1
+ self.attach_metadata.append(kwargs.get("metadata"))
return self._attach_ops
def ReleaseExecute(self, req: proto.ReleaseExecuteRequest, *args,
**kwargs):
+ self.release_metadata.append(kwargs.get("metadata"))
if req.HasField("release_all"):
self.release_calls += 1
elif req.HasField("release_until"):
@@ -149,6 +155,7 @@ if should_test_connect:
def __init__(self, session_id: str, operation_statuses=None):
self._session_id = session_id
self.req = None
+ self.metadata = None
self.client_user_context_extensions = []
if operation_statuses is None:
operation_statuses = self.DEFAULT_OPERATION_STATUSES
@@ -156,6 +163,7 @@ if should_test_connect:
def ExecutePlan(self, req: proto.ExecutePlanRequest, metadata,
timeout=None):
self.req = req
+ self.metadata = metadata
self.client_user_context_extensions =
list(req.user_context.extensions)
resp = proto.ExecutePlanResponse()
resp.session_id = self._session_id
@@ -297,6 +305,30 @@ class SparkConnectClientTestCase(unittest.TestCase):
self.assertEqual(client._user_id, "abc")
+ def test_channel_builder_metadata_is_filtered_per_call(self):
+ class CustomChannelBuilder(DefaultChannelBuilder):
+ def __init__(self):
+ super().__init__("sc://foo/")
+ self.metadata_calls = 0
+
+ def metadata(self):
+ self.metadata_calls += 1
+ return iter(
+ [
+ ("authorization", f"token-{self.metadata_calls}"),
+ ("spark-connect-operation-id", "ignored"),
+ ]
+ )
+
+ builder = CustomChannelBuilder()
+ client = SparkConnectClient(builder, use_reattachable_execute=False)
+ try:
+ self.assertEqual(client._artifact_manager._metadata,
[("authorization", "token-1")])
+ self.assertEqual(client._builder_metadata(), [("authorization",
"token-2")])
+ self.assertEqual(client._builder_metadata(), [("authorization",
"token-3")])
+ finally:
+ client.close()
+
def test_user_context_extension(self):
client = SparkConnectClient("sc://foo/",
use_reattachable_execute=False)
mock = MockService(client._session_id)
@@ -554,6 +586,21 @@ class SparkConnectClientTestCase(unittest.TestCase):
finally:
client.close()
+ def test_execute_plan_sends_operation_id_metadata(self):
+ client = SparkConnectClient(
+ "sc://foo/;spark-connect-operation-id=ignored",
use_reattachable_execute=False
+ )
+ mock = MockService(client._session_id)
+ client._stub = mock
+ try:
+ req = client._execute_plan_request_with_metadata()
+ client._execute(req)
+ self.assertIsNotNone(mock.metadata)
+ values = [v for k, v in mock.metadata if k ==
"spark-connect-operation-id"]
+ self.assertEqual(values, [req.operation_id])
+ finally:
+ client.close()
+
def test_on_exit_calls_release_and_close_when_enabled(self):
client = SparkConnectClient("sc://foo/",
use_reattachable_execute=False)
client._release_session_on_exit = True
@@ -954,6 +1001,21 @@ class
SparkConnectClientReattachTestCase(unittest.TestCase):
eventually(timeout=1, catch_assertions=True)(check_all)()
+ def test_operation_id_metadata_is_sent_on_all_rpcs(self):
+ metadata = [("spark-connect-operation-id", "operation-id")]
+ stub = self._stub_with([self.response], [self.finished])
+ ite = ExecutePlanResponseReattachableIterator(self.request, stub,
self.retrying, metadata)
+ for _ in ite:
+ pass
+
+ def check_all():
+ self.assertEqual(stub.execute_metadata, [metadata])
+ self.assertEqual(stub.attach_metadata, [metadata])
+ self.assertTrue(stub.release_metadata)
+ self.assertTrue(all(value == metadata for value in
stub.release_metadata))
+
+ eventually(timeout=1, catch_assertions=True)(check_all)()
+
def test_fail_during_execute(self):
def fatal():
raise TestException("Fatal")
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 49ae6ed8ea72..6a456a554c5c 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
@@ -45,8 +45,11 @@ class SparkConnectClientSuite extends ConnectFunSuite {
private var service: DummySparkConnectService = _
private var server: Server = _
- private def startDummyServer(port: Int, interceptors: Seq[ServerInterceptor]
= Seq()): Unit = {
- service = new DummySparkConnectService
+ private def startDummyServer(
+ port: Int,
+ interceptors: Seq[ServerInterceptor] = Seq(),
+ dummyService: DummySparkConnectService = new DummySparkConnectService):
Unit = {
+ service = dummyService
val serverBuilder = NettyServerBuilder
.forPort(port)
.addService(service)
@@ -78,19 +81,84 @@ class SparkConnectClientSuite extends ConnectFunSuite {
assert(client.userId == System.getProperty("user.name"))
}
- test("client generates an operation ID for ExecutePlan requests") {
- startDummyServer(0)
- client = SparkConnectClient
- .builder()
- .connectionString(s"sc://localhost:${server.getPort}")
- .disableReattachableExecute()
- .build()
-
- val responses = client.execute(buildPlan("select 1")).toSeq
- val operationId = responses.head.getOperationId
+ Seq(false, true).foreach { reattachable =>
+ test(s"client generates an operation ID for ExecutePlan requests
($reattachable)") {
+ val operationIdHeaders = mutable.Map.empty[String, Option[String]]
+ val interceptor = new ServerInterceptor {
+ override def interceptCall[ReqT, RespT](
+ call: ServerCall[ReqT, RespT],
+ headers: Metadata,
+ next: ServerCallHandler[ReqT, RespT]): ServerCall.Listener[ReqT] =
{
+ val key = Metadata.Key.of(
+ SparkConnectClient.OPERATION_ID_HEADER,
+ Metadata.ASCII_STRING_MARSHALLER)
+ operationIdHeaders.synchronized {
+ operationIdHeaders(call.getMethodDescriptor.getBareMethodName) =
+ Option(headers.get(key))
+ }
+ next.startCall(call, headers)
+ }
+ }
+ val dummyService = if (reattachable) {
+ new DummySparkConnectService {
+ override def executePlan(
+ request: ExecutePlanRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse]): Unit = {
+ responseObserver.onNext(
+ ExecutePlanResponse
+ .newBuilder()
+ .setSessionId(request.getSessionId)
+ .setOperationId(request.getOperationId)
+ .setResponseId("initial-response")
+ .build())
+ responseObserver.onCompleted()
+ }
- UUID.fromString(operationId)
- assert(responses.forall(_.getOperationId == operationId))
+ override def reattachExecute(
+ request: proto.ReattachExecuteRequest,
+ responseObserver: StreamObserver[ExecutePlanResponse]): Unit = {
+ responseObserver.onNext(
+ ExecutePlanResponse
+ .newBuilder()
+ .setSessionId(request.getSessionId)
+ .setOperationId(request.getOperationId)
+ .setResponseId("result-complete")
+
.setResultComplete(proto.ExecutePlanResponse.ResultComplete.newBuilder().build())
+ .build())
+ responseObserver.onCompleted()
+ }
+ }
+ } else {
+ new DummySparkConnectService
+ }
+ startDummyServer(0, Seq(interceptor), dummyService)
+ val builder = SparkConnectClient
+ .builder()
+ .connectionString(s"sc://localhost:${server.getPort}")
+ .option(SparkConnectClient.OPERATION_ID_HEADER, "ignored")
+ if (reattachable) builder.enableReattachableExecute()
+ else builder.disableReattachableExecute()
+ client = builder.build()
+
+ val responses = client.execute(buildPlan("select 1")).toSeq
+ val operationId = responses.head.getOperationId
+
+ UUID.fromString(operationId)
+ assert(responses.forall(_.getOperationId == operationId))
+ assert(operationIdHeaders.synchronized {
+ operationIdHeaders("ExecutePlan").contains(operationId)
+ })
+ if (reattachable) {
+ assert(operationIdHeaders.synchronized {
+ operationIdHeaders("ReattachExecute").contains(operationId)
+ })
+ Eventually.eventually(timeout(5.seconds)) {
+ assert(operationIdHeaders.synchronized {
+ operationIdHeaders("ReleaseExecute").contains(operationId)
+ })
+ }
+ }
+ }
}
test("ExecutePlan exceptions expose the client-generated operation ID") {
diff --git
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/CustomSparkConnectBlockingStub.scala
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/CustomSparkConnectBlockingStub.scala
index 4a1e1c838aa8..688029f4883a 100644
---
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/CustomSparkConnectBlockingStub.scala
+++
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/CustomSparkConnectBlockingStub.scala
@@ -45,10 +45,22 @@ private[connect] class CustomSparkConnectBlockingStub(
// GrpcExceptionConverter with a GRPC stub for fetching error details from
server.
private val grpcExceptionConverter = stubState.exceptionConverter
+ private def executePlanStub(operationId: String) = {
+ Option(operationId)
+ .filter(_.nonEmpty)
+ .map { id =>
+ stub.withInterceptors(
+ new SparkConnectClient.MetadataHeaderClientInterceptor(
+ Map(SparkConnectClient.OPERATION_ID_HEADER -> id)))
+ }
+ .getOrElse(stub)
+ }
+
// Non-reattachable executePlan intentionally has no deadline: a timeout
here would kill the
// server-side execution with no way to recover (there is no ReattachExecute
for this path).
// Use reattachable execution for long-running queries that need deadline
protection.
def executePlan(request: ExecutePlanRequest):
CloseableIterator[ExecutePlanResponse] = {
+ val stubWithOperationId = executePlanStub(request.getOperationId)
grpcExceptionConverter.convert(
request.getSessionId,
request.getUserContext,
@@ -62,7 +74,7 @@ private[connect] class CustomSparkConnectBlockingStub(
request,
r => {
stubState.responseValidator.wrapIterator(
- CloseableIterator(stub.executePlan(r).asScala))
+ CloseableIterator(stubWithOperationId.executePlan(r).asScala))
}),
Option(request.getOperationId).filter(_.nonEmpty))
}
diff --git
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala
index e6428270c20b..c7073c96aba6 100644
---
a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala
+++
b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala
@@ -81,8 +81,14 @@ class ExecutePlanResponseReattachableIterator(
// - this does it's own custom retry handling
// - error conversion is wrapped around this in
CustomSparkConnectBlockingStub,
// this needs raw GRPC errors for retries.
- private val rawBlockingStub =
proto.SparkConnectServiceGrpc.newBlockingStub(channel)
- private val rawAsyncStub = proto.SparkConnectServiceGrpc.newStub(channel)
+ private val operationIdInterceptor = new
SparkConnectClient.MetadataHeaderClientInterceptor(
+ Map(SparkConnectClient.OPERATION_ID_HEADER -> operationId))
+ private val rawBlockingStub =
+ proto.SparkConnectServiceGrpc
+ .newBlockingStub(channel)
+ .withInterceptors(operationIdInterceptor)
+ private val rawAsyncStub =
+
proto.SparkConnectServiceGrpc.newStub(channel).withInterceptors(operationIdInterceptor)
private def stubWithDeadline(deadline: Option[FiniteDuration])
: proto.SparkConnectServiceGrpc.SparkConnectServiceBlockingStub =
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 58ca37fc7c07..5a3bd0b892c5 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
@@ -56,6 +56,14 @@ private[sql] class SparkConnectClient(
private val userContext: UserContext = configuration.userContext
private[this] val stubState = new SparkConnectStubState(channel,
configuration)
+
+ if (configuration.metadata.keys.exists(
+ _.equalsIgnoreCase(SparkConnectClient.OPERATION_ID_HEADER))) {
+ logWarning(
+ s"Connection option ${SparkConnectClient.OPERATION_ID_HEADER} is ignored
because " +
+ "Spark Connect sets it for each ExecutePlan request.")
+ }
+
private[this] val bstub =
new CustomSparkConnectBlockingStub(channel, stubState)
private[this] val stub =
@@ -703,6 +711,8 @@ private final class SparkConnectOperationIdException(val
operationId: String)
object SparkConnectClient {
+ private[connect] val OPERATION_ID_HEADER = "spark-connect-operation-id"
+
/**
* Returns the ExecutePlan operation ID attached to a Spark Connect failure,
when available.
*
@@ -1178,9 +1188,11 @@ object SparkConnectClient {
// Workaround LocalChannelCredentials are added in
// https://github.com/grpc/grpc-java/issues/9900
- var metadataWithOptionalToken = metadata
+ var metadataWithOptionalToken = metadata.filterNot { case (key, _) =>
+ key.equalsIgnoreCase(OPERATION_ID_HEADER)
+ }
if (!isSslEnabled.contains(true) && isLocal && token.isDefined) {
- metadataWithOptionalToken = metadata + (("Authorization", s"Bearer
${token.get}"))
+ metadataWithOptionalToken += (("Authorization", s"Bearer
${token.get}"))
}
if (metadataWithOptionalToken.nonEmpty) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
