This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new c9c45a67b449 [SPARK-57941][CONNECT] Expose Spark Connect sessions and
operations via REST API
c9c45a67b449 is described below
commit c9c45a67b449181a2f26e700ba5ad2c24bbe294c
Author: Venkata Krishnan Sowrirajan <[email protected]>
AuthorDate: Fri Jul 17 15:30:35 2026 -0700
[SPARK-57941][CONNECT] Expose Spark Connect sessions and operations via
REST API
### What changes were proposed in this pull request?
Spark Connect server session and operation (execution) data is collected by
`SparkConnectServerListener` and shown in the "Connect" Spark UI tab, but
unlike SQL ([SPARK-27142](https://issues.apache.org/jira/browse/SPARK-27142))
and Streaming, it is not reachable through the `/api/v1` REST API.
This adds a REST resource mirroring the SQL REST API. New endpoints:
- `GET /applications/[app-id]/connect/sessions` — list sessions (supports
`?offset=&length=`)
- `GET /applications/[app-id]/connect/sessions/[session-id]` — one session
- `GET /applications/[app-id]/connect/operations` — list operations
(supports `?offset=&length=`)
- `GET /applications/[app-id]/connect/operations/detail?jobTag=[job-tag]` —
one operation
(and the `[attempt-id]` variants).
Implementation notes for reviewers:
- The resource classes live in package
`org.apache.spark.status.api.v1.connect`, so they are auto-discovered by the
existing Jersey package scan (`ApiRootResource` registers
`org.apache.spark.status.api.v1` recursively) exactly like the SQL
(`...v1.sql`) and Streaming (`...v1.streaming`) resources. No wiring changes
are needed.
- `ConnectResource` reads from `ui.store.store` — the same `KVStore` that
backs the Connect UI tab — via the existing `SparkConnectServerAppStatusStore`.
It adds no new data collection.
- The operation-detail endpoint takes the `jobTag` as a query parameter.
The `jobTag` is the store's unique natural key `(userId, sessionId,
operationId)`; it embeds the raw user id, which may contain characters (e.g.
`/`) that are not safe in a URL path segment, so it cannot be a path parameter.
The `operationId` alone is not a safe key: it is only unique within a session,
and clients may supply any UUID, so two sessions can share one. Keying on the
natural key via `store.read` also [...]
- The list endpoints push `offset`/`length` into the `KVStoreView`
(`skip`/`max`), matching `SQLAppStatusStore`.
- `SessionInfo`, `ExecutionInfo` and `ExecutionState` are widened from
`private[connect]` to `private[spark]` so the resource package can read them.
They remain internal (not public API).
- The public API surface is the new `SessionData` / `ExecutionData` DTOs in
`api.scala`, following the SQL/Streaming convention (no `since` on these status
DTOs).
Because the Connect listener already persists its events to the event log,
these endpoints also work in the History Server through the existing
`SparkConnectServerHistoryServerPlugin`.
### Why are the changes needed?
The Connect UI tab exposes per-session and per-operation information (user,
timings, state, job/SQL ids, errors) that is only reachable by scraping HTML.
Jobs, stages, SQL and Streaming are all queryable as JSON under `/api/v1`;
Spark Connect should be too, so operators can build monitoring and automation
on top of the Connect server the same way they do for the rest of Spark.
### Does this PR introduce _any_ user-facing change?
Yes. It adds new read-only REST endpoints under
`/api/v1/applications/[app-id]/connect/...` (listed above) and documents them
in `docs/monitoring.md`. There is no change to existing endpoints or behavior.
### How was this patch tested?
New tests added, run with `build/sbt 'connect/testOnly
org.apache.spark.status.api.v1.connect.*'`:
- `ConnectResourceSuite` — unit tests for the `SessionData` /
`ExecutionData` mapping (exercises the private `prepare*` methods directly,
mirroring `SqlResourceSuite`), including numeric ordering of job/SQL ids and a
store lookup that covers the duplicate-`operationId`-across-sessions case.
- `ConnectResourceWithActualDataSuite` — end-to-end test that boots the
real `SparkConnectService` with the UI enabled, runs a query through a Connect
session so the listener populates the KVStore, then hits the endpoints over
HTTP and asserts on the JSON (including detail lookup for a user id containing
`/`, `offset`/`length` paging, and 404s for unknown ids).
```
[info] ConnectResourceSuite:
[info] - prepareSessionData maps SessionInfo fields
[info] - prepareExecutionData maps ExecutionInfo fields and derived
durations
[info] - SPARK-57941: getExecution looks up by the unique job tag
[info] ConnectResourceWithActualDataSuite:
[info] - Connect REST API exposes sessions and operations
[info] - SPARK-57941: operation with a user id containing '/' is fetchable
via the jobTag query
[info] - Connect REST API returns 404 for unknown ids
[info] - SPARK-57941: operations list honors offset and length
[info] Run completed in 12 seconds, 104 milliseconds.
[info] Total number of tests run: 7
[info] Suites: completed 2, aborted 0
[info] Tests: succeeded 7, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
Manual verification against a running Connect server: started the server,
ran `spark.sql("select 1 + 1")` from a Connect client, and confirmed the
endpoints return the expected JSON:
```
$ curl -s .../api/v1/applications/$APPID/connect/sessions
[ {
"sessionId" : "f7901b8a-6aff-4178-9f4c-b6954f172920",
"userId" : "vsowrira",
"startTimestamp" : 1782966263644,
"finishTimestamp" : 0,
"totalExecution" : 2,
"totalTime" : 12375
} ]
$ curl -s .../api/v1/applications/$APPID/connect/operations
[ {
"jobTag" : "SparkConnect_OperationTag_User_..._Operation_...",
"operationId" : "2f8c3023-4c44-4642-bdb1-287324efd702",
"sessionId" : "f7901b8a-6aff-4178-9f4c-b6954f172920",
"userId" : "vsowrira",
"statement" : "sql_command {\n input {\n ... query: \"select 1 + 1 as
two\" ...\n }\n}\n",
"state" : "CLOSED",
"startTimestamp" : 1782966263648,
"finishTimestamp" : 1782966264286,
"closeTimestamp" : 1782966264301,
"duration" : 653,
"executionTime" : 638,
"sparkSessionTags" : [ ],
"jobIds" : [ ],
"sqlExecIds" : [ ],
"detail" : ""
}, ... ]
```
scalafmt check passes for the connect module (`./build/mvn scalafmt:format
-Dscalafmt.validateOnly=true -pl sql/connect/server` → `BUILD SUCCESS`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57016 from venkata91/connect-rest-api.
Authored-by: Venkata Krishnan Sowrirajan <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
---
docs/monitoring.md | 22 +++
.../ui/SparkConnectServerAppStatusStore.scala | 14 +-
.../api/v1/connect/ApiConnectRootResource.scala | 36 +++++
.../status/api/v1/connect/ConnectResource.scala | 101 +++++++++++++
.../apache/spark/status/api/v1/connect/api.scala | 42 ++++++
.../api/v1/connect/ConnectResourceSuite.scala | 148 ++++++++++++++++++
.../ConnectResourceWithActualDataSuite.scala | 165 +++++++++++++++++++++
7 files changed, 525 insertions(+), 3 deletions(-)
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 4f90cbb5645c..0c178110e8f6 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -714,6 +714,28 @@ can be identified by their `[attempt-id]`. In the API
listed below, when running
<code>?planDescription=[true (default) | false]</code> enables/disables
Physical <code>planDescription</code> on demand for the given query when
Physical Plan size is high.
</td>
</tr>
+ <tr>
+ <td><code>/applications/[app-id]/connect/sessions</code></td>
+ <td>A list of all Spark Connect sessions for a given application.
+ <br>
+ <code>?offset=[offset]&length=[len]</code> lists sessions in the given
range.
+ </td>
+ </tr>
+ <tr>
+ <td><code>/applications/[app-id]/connect/sessions/[session-id]</code></td>
+ <td>Details for the given Spark Connect session.</td>
+ </tr>
+ <tr>
+ <td><code>/applications/[app-id]/connect/operations</code></td>
+ <td>A list of all Spark Connect operations for a given application.
+ <br>
+ <code>?offset=[offset]&length=[len]</code> lists operations in the given
range.
+ </td>
+ </tr>
+ <tr>
+
<td><code>/applications/[app-id]/connect/operations/detail?jobTag=[job-tag]</code></td>
+ <td>Details for the Spark Connect operation with the given job tag.</td>
+ </tr>
<tr>
<td><code>/applications/[app-id]/environment</code></td>
<td>Environment details of the given application.</td>
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerAppStatusStore.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerAppStatusStore.scala
index 28812e49d1fa..b5211b6c3072 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerAppStatusStore.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerAppStatusStore.scala
@@ -31,10 +31,18 @@ class SparkConnectServerAppStatusStore(store: KVStore) {
KVUtils.viewToSeq(store.view(classOf[SessionInfo]))
}
+ def getSessionList(offset: Int, length: Int): Seq[SessionInfo] = {
+
KVUtils.viewToSeq(store.view(classOf[SessionInfo]).skip(offset).max(length))
+ }
+
def getExecutionList: Seq[ExecutionInfo] = {
KVUtils.viewToSeq(store.view(classOf[ExecutionInfo]))
}
+ def getExecutionList(offset: Int, length: Int): Seq[ExecutionInfo] = {
+
KVUtils.viewToSeq(store.view(classOf[ExecutionInfo]).skip(offset).max(length))
+ }
+
def getOnlineSessionNum: Int = {
KVUtils.count(store.view(classOf[SessionInfo]))(_.finishTimestamp == 0)
}
@@ -73,7 +81,7 @@ class SparkConnectServerAppStatusStore(store: KVStore) {
}
}
-private[connect] class SessionInfo(
+private[spark] class SessionInfo(
@KVIndexParam val sessionId: String,
val startTimestamp: Long,
val userId: String,
@@ -90,7 +98,7 @@ private[connect] class SessionInfo(
}
}
-private[connect] class ExecutionInfo(
+private[spark] class ExecutionInfo(
@KVIndexParam val jobTag: String,
val statement: String,
val sessionId: String,
@@ -125,7 +133,7 @@ private[connect] class ExecutionInfo(
}
}
-private[connect] object ExecutionState extends Enumeration {
+private[spark] object ExecutionState extends Enumeration {
val STARTED, COMPILED, READY, CANCELED, FAILED, FINISHED, CLOSED = Value
type ExecutionState = Value
}
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ApiConnectRootResource.scala
b/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ApiConnectRootResource.scala
new file mode 100644
index 000000000000..7a19712d8bde
--- /dev/null
+++
b/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ApiConnectRootResource.scala
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.status.api.v1.connect
+
+import jakarta.ws.rs.{Path, PathParam}
+
+import org.apache.spark.status.api.v1.ApiRequestContext
+
+@Path("/v1")
+private[v1] class ApiConnectRootResource extends ApiRequestContext {
+
+ @Path("applications/{appId}/connect")
+ def connectList(@PathParam("appId") appId: String): Class[ConnectResource] =
+ classOf[ConnectResource]
+
+ @Path("applications/{appId}/{attemptId}/connect")
+ def connectList(
+ @PathParam("appId") appId: String,
+ @PathParam("attemptId") attemptId: String): Class[ConnectResource] =
+ classOf[ConnectResource]
+}
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ConnectResource.scala
b/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ConnectResource.scala
new file mode 100644
index 000000000000..6e85ab5ead2b
--- /dev/null
+++
b/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/ConnectResource.scala
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.status.api.v1.connect
+
+import jakarta.ws.rs._
+import jakarta.ws.rs.core.MediaType
+
+import org.apache.spark.sql.connect.ui.{ExecutionInfo, SessionInfo,
SparkConnectServerAppStatusStore}
+import org.apache.spark.status.api.v1.{BadParameterException, BaseAppResource,
NotFoundException}
+
+@Produces(Array(MediaType.APPLICATION_JSON))
+private[v1] class ConnectResource extends BaseAppResource {
+
+ @GET
+ @Path("sessions")
+ def sessionList(
+ @DefaultValue("0") @QueryParam("offset") offset: Int,
+ @DefaultValue("-1") @QueryParam("length") length: Int): Seq[SessionData]
= withUI { ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ val sessions =
+ if (length <= 0) store.getSessionList else store.getSessionList(offset,
length)
+ sessions.map(prepareSessionData)
+ }
+
+ @GET
+ @Path("sessions/{sessionId}")
+ def session(@PathParam("sessionId") sessionId: String): SessionData = withUI
{ ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ store
+ .getSession(sessionId)
+ .map(prepareSessionData)
+ .getOrElse(throw new NotFoundException("unknown session id: " +
sessionId))
+ }
+
+ @GET
+ @Path("operations")
+ def operationList(
+ @DefaultValue("0") @QueryParam("offset") offset: Int,
+ @DefaultValue("-1") @QueryParam("length") length: Int):
Seq[ExecutionData] = withUI { ui =>
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ val operations =
+ if (length <= 0) store.getExecutionList else
store.getExecutionList(offset, length)
+ operations.map(prepareExecutionData)
+ }
+
+ // The job tag is the unique natural key (userId, sessionId, operationId).
It is taken as a query
+ // parameter rather than a path segment because it embeds the raw user id,
which may contain
+ // characters such as '/' that are not safe in a path segment.
+ @GET
+ @Path("operations/detail")
+ def operation(@QueryParam("jobTag") jobTag: String): ExecutionData = withUI
{ ui =>
+ if (jobTag == null || jobTag.isEmpty) {
+ throw new BadParameterException("jobTag is required.")
+ }
+ val store = new SparkConnectServerAppStatusStore(ui.store.store)
+ store
+ .getExecution(jobTag)
+ .map(prepareExecutionData)
+ .getOrElse(throw new NotFoundException("unknown jobTag: " + jobTag))
+ }
+
+ private def prepareSessionData(info: SessionInfo): SessionData = new
SessionData(
+ sessionId = info.sessionId,
+ userId = info.userId,
+ startTimestamp = info.startTimestamp,
+ finishTimestamp = info.finishTimestamp,
+ totalExecution = info.totalExecution,
+ totalTime = info.totalTime)
+
+ private def prepareExecutionData(info: ExecutionInfo): ExecutionData = new
ExecutionData(
+ jobTag = info.jobTag,
+ operationId = info.operationId,
+ sessionId = info.sessionId,
+ userId = info.userId,
+ statement = info.statement,
+ state = info.state.toString,
+ startTimestamp = info.startTimestamp,
+ finishTimestamp = info.finishTimestamp,
+ closeTimestamp = info.closeTimestamp,
+ duration = info.totalTime(info.closeTimestamp),
+ executionTime = info.totalTime(info.finishTimestamp),
+ sparkSessionTags = info.sparkSessionTags.toSeq.sorted,
+ jobIds = info.jobId.toSeq.sortBy(_.toInt),
+ sqlExecIds = info.sqlExecId.toSeq.sortBy(_.toLong),
+ detail = info.detail)
+}
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/api.scala
b/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/api.scala
new file mode 100644
index 000000000000..36c87164029c
--- /dev/null
+++
b/sql/connect/server/src/main/scala/org/apache/spark/status/api/v1/connect/api.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.status.api.v1.connect
+
+class SessionData private[spark] (
+ val sessionId: String,
+ val userId: String,
+ val startTimestamp: Long,
+ val finishTimestamp: Long,
+ val totalExecution: Long,
+ val totalTime: Long)
+
+class ExecutionData private[spark] (
+ val jobTag: String,
+ val operationId: String,
+ val sessionId: String,
+ val userId: String,
+ val statement: String,
+ val state: String,
+ val startTimestamp: Long,
+ val finishTimestamp: Long,
+ val closeTimestamp: Long,
+ val duration: Long,
+ val executionTime: Long,
+ val sparkSessionTags: Seq[String],
+ val jobIds: Seq[String],
+ val sqlExecIds: Seq[String],
+ val detail: String)
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/status/api/v1/connect/ConnectResourceSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/status/api/v1/connect/ConnectResourceSuite.scala
new file mode 100644
index 000000000000..253ab84f2a62
--- /dev/null
+++
b/sql/connect/server/src/test/scala/org/apache/spark/status/api/v1/connect/ConnectResourceSuite.scala
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.status.api.v1.connect
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+import org.scalatest.PrivateMethodTester
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.internal.config.Status.{ASYNC_TRACKING_ENABLED,
LIVE_ENTITY_UPDATE_PERIOD}
+import org.apache.spark.sql.connect.service.ExecuteJobTag
+import org.apache.spark.sql.connect.ui.{ExecutionInfo, ExecutionState,
SessionInfo, SparkConnectServerAppStatusStore}
+import org.apache.spark.status.ElementTrackingStore
+import org.apache.spark.util.kvstore.InMemoryStore
+
+/**
+ * Unit tests for the Spark Connect REST API resource's data mapping. Mirrors
SqlResourceSuite:
+ * the private prepare* methods are exercised directly so no servlet context /
live SparkUI is
+ * required.
+ */
+class ConnectResourceSuite extends SparkFunSuite with PrivateMethodTester {
+
+ private val resource = new ConnectResource()
+ private val prepareSessionData =
PrivateMethod[SessionData](Symbol("prepareSessionData"))
+ private val prepareExecutionData =
PrivateMethod[ExecutionData](Symbol("prepareExecutionData"))
+
+ test("prepareSessionData maps SessionInfo fields") {
+ val info = new SessionInfo(
+ sessionId = "session-1",
+ startTimestamp = 1000L,
+ userId = "user-1",
+ finishTimestamp = 3000L,
+ totalExecution = 5L)
+
+ val data = resource invokePrivate prepareSessionData(info)
+
+ assert(data.sessionId === "session-1")
+ assert(data.userId === "user-1")
+ assert(data.startTimestamp === 1000L)
+ assert(data.finishTimestamp === 3000L)
+ assert(data.totalExecution === 5L)
+ assert(data.totalTime === 2000L)
+ }
+
+ test("prepareExecutionData maps ExecutionInfo fields and derived durations")
{
+ val info = new ExecutionInfo(
+ jobTag = "job-tag-1",
+ statement = "SELECT 1",
+ sessionId = "session-1",
+ startTimestamp = 1000L,
+ userId = "user-1",
+ operationId = "op-1",
+ sparkSessionTags = Set("tagB", "tagA"),
+ finishTimestamp = 2000L,
+ closeTimestamp = 3000L,
+ detail = "",
+ state = ExecutionState.CLOSED,
+ jobId = ArrayBuffer("10", "2"),
+ // A SQL execution id can exceed Int.MaxValue (it comes from a
process-wide AtomicLong).
+ sqlExecId = mutable.Set("4294967296", "10"))
+
+ val data = resource invokePrivate prepareExecutionData(info)
+
+ assert(data.jobTag === "job-tag-1")
+ assert(data.operationId === "op-1")
+ assert(data.sessionId === "session-1")
+ assert(data.userId === "user-1")
+ assert(data.statement === "SELECT 1")
+ assert(data.state === "CLOSED")
+ assert(data.startTimestamp === 1000L)
+ assert(data.finishTimestamp === 2000L)
+ assert(data.closeTimestamp === 3000L)
+ assert(data.duration === 2000L) // closeTimestamp - startTimestamp
+ assert(data.executionTime === 1000L) // finishTimestamp - startTimestamp
+ assert(data.sparkSessionTags === Seq("tagA", "tagB"))
+ // Numeric ids stored as strings must sort numerically, not
lexicographically ("2" before "10").
+ // SQL execution ids are Long-valued, so they must not be parsed as Int.
+ assert(data.jobIds === Seq("2", "10"))
+ assert(data.sqlExecIds === Seq("10", "4294967296"))
+ assert(data.detail === "")
+ }
+
+ test("SPARK-57941: getExecution looks up by the unique job tag") {
+ // Backs the operation-detail endpoint, and covers the store the History
Server populates when
+ // it replays events into a fresh KVStore. The job tag is the unique
natural key; two sessions
+ // may legally share an operationId UUID, so the lookup must key on the
full identity.
+ val conf = new SparkConf()
+ .set(ASYNC_TRACKING_ENABLED, false)
+ .set(LIVE_ENTITY_UPDATE_PERIOD, 0L)
+ val kvstore = new ElementTrackingStore(new InMemoryStore, conf)
+ try {
+ val store = new SparkConnectServerAppStatusStore(kvstore)
+ val userId = "tenant/alice"
+ val operationId = "00112233-4455-6677-8899-aabbccddeeff"
+ // Same operationId, different sessions -> distinct executions with
distinct job tags.
+ val jobTag1 = ExecuteJobTag(userId, "session-1", operationId)
+ val jobTag2 = ExecuteJobTag(userId, "session-2", operationId)
+ kvstore.write(newExecutionInfo(jobTag1, "session-1", userId,
operationId))
+ kvstore.write(newExecutionInfo(jobTag2, "session-2", userId,
operationId))
+
+ val found1 = store.getExecution(jobTag1)
+ assert(found1.isDefined, "execution should be retrievable by its job
tag")
+ assert(found1.get.jobTag === jobTag1)
+ assert(found1.get.sessionId === "session-1")
+ assert(found1.get.jobTag.contains(userId), "job tag embeds the raw user
id")
+ // The shared operationId must not collapse the two rows onto each other.
+
assert(store.getExecution(jobTag2).map(_.sessionId).contains("session-2"))
+ assert(store.getExecution("no-such-tag").isEmpty)
+ } finally {
+ kvstore.close()
+ }
+ }
+
+ private def newExecutionInfo(
+ jobTag: String,
+ sessionId: String,
+ userId: String,
+ operationId: String): ExecutionInfo = new ExecutionInfo(
+ jobTag = jobTag,
+ statement = "SELECT 1",
+ sessionId = sessionId,
+ startTimestamp = 1000L,
+ userId = userId,
+ operationId = operationId,
+ sparkSessionTags = Set.empty,
+ finishTimestamp = 2000L,
+ closeTimestamp = 3000L,
+ detail = "",
+ state = ExecutionState.CLOSED,
+ jobId = ArrayBuffer.empty,
+ sqlExecId = mutable.Set.empty)
+}
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/status/api/v1/connect/ConnectResourceWithActualDataSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/status/api/v1/connect/ConnectResourceWithActualDataSuite.scala
new file mode 100644
index 000000000000..2b3c9344b1de
--- /dev/null
+++
b/sql/connect/server/src/test/scala/org/apache/spark/status/api/v1/connect/ConnectResourceWithActualDataSuite.scala
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.status.api.v1.connect
+
+import java.io.IOException
+import java.net.{HttpURLConnection, URI, URL}
+
+import org.json4s.{DefaultFormats, Formats}
+import org.json4s.jackson.JsonMethods
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.connect.SparkConnectServerTest
+import org.apache.spark.sql.connect.ui.ExecutionState
+import org.apache.spark.util.Utils
+
+/**
+ * End-to-end test for the Spark Connect REST API: boots the real
SparkConnectService with the UI
+ * enabled, runs a real query through a Connect session so the listener
populates the KVStore,
+ * then exercises the `/api/v1/applications/{appId}/connect/...` endpoints
over HTTP. Mirrors
+ * SqlResourceWithActualMetricsSuite.
+ */
+class ConnectResourceWithActualDataSuite extends SparkConnectServerTest {
+
+ private implicit val formats: Formats = DefaultFormats
+
+ // Populated per JSON payload; only the fields we assert on are declared.
json4s ignores the rest.
+ private case class TestSessionData(sessionId: String, userId: String,
totalExecution: Long)
+ private case class TestExecutionData(
+ jobTag: String,
+ operationId: String,
+ sessionId: String,
+ userId: String,
+ state: String)
+
+ override def sparkConf: SparkConf = super.sparkConf.set("spark.ui.enabled",
"true")
+
+ private def baseUrl: String =
+ spark.sparkContext.ui.get.webUrl +
+ s"/api/v1/applications/${spark.sparkContext.applicationId}/connect"
+
+ private def get(url: URL): (Int, Option[String]) = {
+ val conn = url.openConnection().asInstanceOf[HttpURLConnection]
+ conn.setRequestMethod("GET")
+ conn.connect()
+ val code = conn.getResponseCode
+ val body =
+ try Option(conn.getInputStream).map(Utils.toString)
+ catch { case _: IOException =>
Option(conn.getErrorStream).map(Utils.toString) }
+ (code, body)
+ }
+
+ private def getOk(path: String): String = {
+ val (code, body) = get(new URI(s"$baseUrl$path").toURL)
+ assert(code == 200, s"expected 200 for $path but got $code,
body=${body.orNull}")
+ body.get
+ }
+
+ private def enc(s: String): String =
+ java.net.URLEncoder.encode(s, java.nio.charset.StandardCharsets.UTF_8)
+
+ test("Connect REST API exposes sessions and operations") {
+ // Run a real query through a Connect session; this posts the
session/operation listener
+ // events that the SparkConnectServerListener writes into the KVStore.
+ withSession { session =>
+ assert(session.sql("select 1 + 1").collect().head.getInt(0) === 2)
+ }
+ // Listener events are posted asynchronously; drain the bus before reading
the store.
+ spark.sparkContext.listenerBus.waitUntilEmpty(eventuallyTimeout.toMillis)
+
+ eventuallyWithTimeout {
+ val sessions =
JsonMethods.parse(getOk("/sessions")).extract[List[TestSessionData]]
+ val session = sessions.find(_.sessionId == defaultSessionId)
+ assert(session.isDefined, s"session $defaultSessionId not in $sessions")
+ assert(session.get.userId === defaultUserId)
+ assert(session.get.totalExecution >= 1L)
+
+ val operations =
JsonMethods.parse(getOk("/operations")).extract[List[TestExecutionData]]
+ val op = operations.find(_.sessionId == defaultSessionId)
+ assert(op.isDefined, s"no operation for session $defaultSessionId in
$operations")
+ assert(op.get.userId === defaultUserId)
+ assert(ExecutionState.values.map(_.toString).contains(op.get.state))
+
+ val one =
JsonMethods.parse(getOk(s"/sessions/$defaultSessionId")).extract[TestSessionData]
+ assert(one.sessionId === defaultSessionId)
+
+ val oneOp = JsonMethods
+ .parse(getOk(s"/operations/detail?jobTag=${enc(op.get.jobTag)}"))
+ .extract[TestExecutionData]
+ assert(oneOp.operationId === op.get.operationId)
+ assert(oneOp.jobTag === op.get.jobTag)
+ }
+ }
+
+ test("SPARK-57941: operation with a user id containing '/' is fetchable via
the jobTag query") {
+ // The job tag embeds the raw user id, so a '/' in it would break a path
segment. The detail
+ // endpoint takes the job tag as a query parameter, which handles such
user ids.
+ val userId = "tenant/alice"
+ val sessionId = java.util.UUID.randomUUID.toString
+ withSession(sessionId = sessionId, userId = userId) { session =>
+ assert(session.sql("select 1 + 1").collect().head.getInt(0) === 2)
+ }
+ spark.sparkContext.listenerBus.waitUntilEmpty(eventuallyTimeout.toMillis)
+
+ eventuallyWithTimeout {
+ val operations =
JsonMethods.parse(getOk("/operations")).extract[List[TestExecutionData]]
+ val op = operations.find(o => o.sessionId == sessionId && o.userId ==
userId)
+ assert(op.isDefined, s"no operation for user $userId in $operations")
+ assert(op.get.jobTag.contains(userId))
+ val oneOp = JsonMethods
+ .parse(getOk(s"/operations/detail?jobTag=${enc(op.get.jobTag)}"))
+ .extract[TestExecutionData]
+ assert(oneOp.jobTag === op.get.jobTag)
+ assert(oneOp.userId === userId)
+ }
+ }
+
+ test("Connect REST API returns 404 for unknown ids") {
+ assert(get(new URI(s"$baseUrl/sessions/does-not-exist").toURL)._1 === 404)
+ assert(get(new
URI(s"$baseUrl/operations/detail?jobTag=does-not-exist").toURL)._1 === 404)
+ }
+
+ test("SPARK-57941: operation detail without a jobTag returns 400") {
+ assert(get(new URI(s"$baseUrl/operations/detail").toURL)._1 === 400)
+ assert(get(new URI(s"$baseUrl/operations/detail?jobTag=").toURL)._1 ===
400)
+ }
+
+ test("SPARK-57941: operations list honors offset and length") {
+ withSession { session =>
+ (1 to 3).foreach(i => assert(session.sql(s"select
$i").collect().head.getInt(0) === i))
+ }
+ spark.sparkContext.listenerBus.waitUntilEmpty(eventuallyTimeout.toMillis)
+
+ eventuallyWithTimeout {
+ val all =
JsonMethods.parse(getOk("/operations")).extract[List[TestExecutionData]]
+ assert(all.size >= 3, s"expected at least 3 operations, got ${all.size}")
+
+ // length bounds the page size...
+ val firstTwo =
+
JsonMethods.parse(getOk("/operations?length=2")).extract[List[TestExecutionData]]
+ assert(firstTwo.size === 2)
+ assert(firstTwo.map(_.jobTag) === all.take(2).map(_.jobTag))
+
+ // ...and offset skips from the front, matching a slice of the full list.
+ val skipOne = JsonMethods
+ .parse(getOk("/operations?offset=1&length=2"))
+ .extract[List[TestExecutionData]]
+ assert(skipOne.map(_.jobTag) === all.slice(1, 3).map(_.jobTag))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
