This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 5029f6761395 [SPARK-58451][CONNECT] Filter null extension elements
returned by GetStatus plugins
5029f6761395 is described below
commit 5029f6761395115fa8b9584188f02483b4e34dac
Author: haoyangeng-db <[email protected]>
AuthorDate: Fri Aug 7 13:52:26 2026 -0700
[SPARK-58451][CONNECT] Filter null extension elements returned by GetStatus
plugins
### What changes were proposed in this pull request?
`SparkConnectGetStatusHandler` collects response extensions from every
registered `GetStatusPlugin` and adds them to the response with
`GetStatusResponse.Builder.addExtensions` /
`OperationStatus.Builder.addExtensions`. Those protobuf builder methods reject
`null`, so a plugin that returns a list containing a `null` element makes the
whole `GetStatus` RPC fail with a NullPointerException that escapes the
per-plugin `try`/`catch` isolation.
This filters `null` elements out inside that isolation boundary, at both
the request level and the operation level, so one misbehaving plugin can no
longer break the RPC for everyone.
### Why are the changes needed?
The handler already goes to some length to isolate plugin failures: each
plugin call is wrapped in `try`/`catch NonFatal`, and a throwing plugin is
logged and skipped so healthy plugins still contribute. A `null` list element
defeats that, because the NullPointerException is raised later by the builder,
outside the `try` block, and fails the request. Plugins are third-party
extension points, so the handler should not assume their returned lists are
null-free.
### Does this PR introduce _any_ user-facing change?
No. With well-behaved plugins the behavior is unchanged; only the failure
mode for a plugin returning a null element changes, from a failed RPC to that
element being dropped.
### How was this patch tested?
Added `GetStatusHandlerSuite."GetStatus filters null extension elements
returned by a plugin"`, which registers a plugin returning a single-null list
alongside a healthy echo plugin and asserts the null is dropped while the
healthy plugin's extension survives, at both the request and operation levels.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored by Claude Code.
Closes #57638 from haoyangeng-db/getstatus-plugin-null-extensions.
Authored-by: haoyangeng-db <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 26b3fe6d79a8f4b8fa40a6ccab15e03b0a98ac5f)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../service/SparkConnectGetStatusHandler.scala | 6 ++-
.../connect/service/GetStatusHandlerSuite.scala | 48 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 2 deletions(-)
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectGetStatusHandler.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectGetStatusHandler.scala
index c37061ff2eba..fcc7089f27aa 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectGetStatusHandler.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectGetStatusHandler.scala
@@ -179,7 +179,8 @@ class SparkConnectGetStatusHandler(responseObserver:
StreamObserver[proto.GetSta
SparkConnectPluginRegistry.getStatusRegistry.flatMap { plugin =>
try {
plugin.processRequestExtensions(sessionHolder,
requestExtensions).toScala match {
- case Some(extensions) => extensions.asScala.toSeq
+ // Filter nulls inside the isolation boundary so a null element
can't NPE addExtensions.
+ case Some(extensions) => extensions.asScala.iterator.filter(_ !=
null).toSeq
case None => Seq.empty
}
} catch {
@@ -202,7 +203,8 @@ class SparkConnectGetStatusHandler(responseObserver:
StreamObserver[proto.GetSta
plugin
.processOperationExtensions(operationId, sessionHolder,
operationExtensions)
.toScala match {
- case Some(extensions) => extensions.asScala.toSeq
+ // Filter nulls inside the isolation boundary so a null element
can't NPE addExtensions.
+ case Some(extensions) => extensions.asScala.iterator.filter(_ !=
null).toSeq
case None => Seq.empty
}
} catch {
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/GetStatusHandlerSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/GetStatusHandlerSuite.scala
index 21d96fd8a87d..e8e960fdb578 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/GetStatusHandlerSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/GetStatusHandlerSuite.scala
@@ -88,6 +88,28 @@ class NoOpGetStatusPlugin extends GetStatusPlugin {
Optional.empty()
}
+/**
+ * A plugin that returns a list containing a null element, to test null
filtering.
+ */
+class NullElementGetStatusPlugin extends GetStatusPlugin {
+ override def processRequestExtensions(
+ sessionHolder: SessionHolder,
+ requestExtensions: util.List[protobuf.Any]):
Optional[util.List[protobuf.Any]] =
+ listWithNull()
+
+ override def processOperationExtensions(
+ operationId: String,
+ sessionHolder: SessionHolder,
+ operationExtensions: util.List[protobuf.Any]):
Optional[util.List[protobuf.Any]] =
+ listWithNull()
+
+ private def listWithNull(): Optional[util.List[protobuf.Any]] = {
+ val result = new util.ArrayList[protobuf.Any]()
+ result.add(null)
+ Optional.of(result)
+ }
+}
+
/**
* A plugin that always throws a RuntimeException.
*/
@@ -344,6 +366,32 @@ class GetStatusHandlerSuite extends SharedSparkSession {
assert(opExtValues.contains(s"op-echo:${executeHolder.operationId}:safe"))
assert(opExtValues.contains(s"second-op:${executeHolder.operationId}:safe"))
}
+
+ test("GetStatus filters null extension elements returned by a plugin") {
+ SparkConnectPluginRegistry.setGetStatusPluginsForTesting(
+ Seq(new NullElementGetStatusPlugin(), new EchoGetStatusPlugin()))
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ val command = proto.Command.newBuilder().build()
+ val executeHolder =
SparkConnectTestUtils.createDummyExecuteHolder(sessionHolder, command)
+
+ val reqExt = protobuf.Any.pack(StringValue.of("data"))
+ val opExt = protobuf.Any.pack(StringValue.of("data"))
+ val response = sendGetOperationStatusRequest(
+ sessionHolder.sessionId,
+ operationIds = Seq(executeHolder.operationId),
+ userId = sessionHolder.userId,
+ requestExtensions = Seq(reqExt),
+ operationExtensions = Seq(opExt))
+
+ // The null element is dropped; only the healthy plugin's extension
survives, at both levels.
+ val responseExtValues = response.getExtensionsList.asScala
+ .map(_.unpack(classOf[StringValue]).getValue)
+ assert(responseExtValues == Seq("request-echo:data"))
+
+ val opExtValues =
response.getOperationStatusesList.asScala.head.getExtensionsList.asScala
+ .map(_.unpack(classOf[StringValue]).getValue)
+ assert(opExtValues == Seq(s"op-echo:${executeHolder.operationId}:data"))
+ }
}
private class GetStatusResponseObserver extends
StreamObserver[proto.GetStatusResponse] {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
