[SPARK-57941][CONNECT] Expose Spark Connect sessions and operations via REST API

[venkata91]

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) 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
• GET /applications/[app-id]/connect/sessions/[session-id] — one session
• GET /applications/[app-id]/connect/operations — list operations
• GET /applications/[app-id]/connect/operations/[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.
• 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).
• 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 404s for unknown ids).

[info] ConnectResourceSuite:
[info] - prepareSessionData maps SessionInfo fields (11 milliseconds)
[info] - prepareExecutionData maps ExecutionInfo fields and derived durations (5 milliseconds)
[info] ConnectResourceWithActualDataSuite:
[info] - Connect REST API exposes sessions and operations (2 seconds, 649 milliseconds)
[info] - Connect REST API returns 404 for unknown ids (11 milliseconds)
[info] Run completed in 6 seconds, 485 milliseconds.
[info] Total number of tests run: 4
[info] Suites: completed 2, aborted 0
[info] Tests: succeeded 4, 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)

[venkata91]

cc @yaooqinn @pan3793 @dongjoon-hyun @cloud-fan @HyukjinKwon for review.

REST API for the Spark Connect server, mirroring the SQL REST API pattern. It reads from the same KVStore that backs the Connect UI tab and should work with the History Server as well. Would appreciate your review when you have a chance. Thanks!

[sunchao]

Summary

Prior state and problem

Spark Connect already records session and operation history for the Connect UI, but operators cannot consume the same data through Spark's /api/v1 interface. This PR fills that gap for both live applications and History Server playback.

Design approach

The change follows the existing SQL status API pattern: Jersey discovers a new Connect resource package, the resource reads the existing Connect KVStore, and public DTOs provide JSON responses. It adds application and application-attempt routes without creating a second listener or collection path.

Correctness / compatibility analysis

The application scoping, view authorization, KVStore lookups, 404 behavior, history reconstruction, duration calculations, deterministic collection serialization, and internal visibility changes are consistent with the surrounding Spark status APIs. I found one P2 in the operation-detail route: jobTag is not guaranteed to be a safe URL path segment, so some operations returned by the list endpoint cannot be fetched individually. The inline comment has the triggering case and suggested fix.

The new DTOs intentionally expose the existing Connect UI/history representation, including millisecond timestamps, 0 sentinels for unfinished events, and CLOSED as the final query-history state. These choices do not break an existing API, but they will become part of the public wire contract.

Key design decisions

• Reuse SparkConnectServerAppStatusStore instead of duplicating state.
• Discover resources recursively under org.apache.spark.status.api.v1, matching SQL and Streaming.
• Keep listener/store classes internal by widening them only to private[spark].
• Return statements, lifecycle state, tags, job IDs, SQL execution IDs, and error details in the operation DTO.

Implementation sketch

ApiConnectRootResource supplies live and attempt-aware routes. ConnectResource maps SessionInfo and ExecutionInfo into SessionData and ExecutionData, while BaseAppResource supplies application lookup and UI authorization. The tests cover field mapping, live HTTP discovery, list/detail retrieval, and missing-ID responses.

Behavioral changes worth calling out

This adds four read-only endpoint families under /applications/[app-id]/connect, plus attempt variants. The responses expose the same redacted statement and error information already available to authorized Connect UI viewers. Existing endpoint behavior and data collection are unchanged.

Suggested improvements

Please address the inline P2 before merging. As optional API hardening, consider adding SQL-style offset / length and reduced-detail controls for operation lists, and adding explicit History Server replay coverage.

Upstream checks are green. The focused mapping suite passed locally; the live-server suite could not initialize under this machine's unsupported JDK 26 because jdk.internal.ref.Cleaner is unavailable, which is an environment limitation rather than a PR failure.

[sunchao]

[P2] Keep opaque job tags out of a single path segment

jobTag embeds the raw Connect userId, and valid user IDs may contain /. Such an operation appears in /operations, but its returned tag cannot be used with this endpoint: a literal slash creates another route segment, while encoding it as %2F is rejected by Spark's default Jetty 12.1 URI compliance as AMBIGUOUS_PATH_SEPARATOR. Please use a path-safe lookup key (for example operationId) or accept the opaque tag as a query parameter, and add regression coverage with a user ID such as tenant/alice.

[venkata91]

Good catch, thanks — fixed. The operation-detail route now keys on operationId (a validated UUID, so always path-safe) instead of the jobTag:

• Added a secondary @KVIndex("operationId") on ExecutionInfo and a getExecutionByOperationId lookup on the store.
• GET .../connect/operations/{operationId} now looks up by that index; jobTag is still returned in the DTO for reference.
• Regression test added with userId = "tenant/alice" (ConnectResourceWithActualDataSuite): it drives a real query, confirms the returned jobTag contains the raw /, and that the operation is still fetchable via its operationId. A store-level test in ConnectResourceSuite exercises the new index directly (also covering the History Server replay path, which uses the same SparkConnectServerAppStatusStore).

Implementation note: I used a method-based @JsonIgnore @KVIndex("operationId") def rather than annotating the constructor val, because Scala targets a param-val annotation at the parameter (not the field/getter), so the KVStore reflection doesn't pick it up — the tests caught that.

[venkata91]

Thanks for the thorough review, @sunchao! Addressed in the latest push (8629e174e53):

• [P2] job tag not path-safe — fixed. The operation-detail endpoint now looks up by operationId (a UUID) via a new secondary KVStore index, instead of the jobTag which embeds the raw user id. Added a tenant/alice regression test proving list→detail works, plus a store-level index test that also covers the History Server replay path.
• Pagination — added SQL-style offset/length query params to both the /sessions and /operations list endpoints, and documented them in monitoring.md.
• Reduced-detail controls — I skipped a details=false toggle here on purpose: unlike the SQL API (whose details skips the heavy plan-graph/metrics payload), the Connect ExecutionData is all lightweight scalar fields, so a toggle would add API surface without meaningfully reducing the response. Happy to add it if you'd prefer symmetry with the SQL API.

All connect REST suites pass locally on the current master (rebased). The DTO wire-contract fields you flagged (ms timestamps, 0 sentinels, CLOSED terminal state) are unchanged and match the existing Connect UI/history representation.

[venkata91]

@sunchao Addressed your review comments. Please take a look again whenever you get a chance. Looks like the failing test is unrelated and flaky. Not sure how to make that green though :)

[vaquarkhan]

jobId/sqlExecId are numeric ids stored as strings, so .sorted is lexicographic: ["2","10"] returns as ["10","2"]. The unit test only passes because "1","2" and "0","10" sort the same both ways. Use .sortBy(_.toInt) (both lines), or document the order is intentionally lexicographic.

[vaquarkhan]

This slices after the caller has already materialized the whole store getSessionList/getExecutionList (lines 35 and 54) call KVUtils.viewToSeq(store.view(...)), loading all N rows on every request. So ?offset=0&length=10 still walks the entire store, and the default length=-1 (lines 33 and 52) means a bare GET /connect/operations returns an unbounded payload an OOM/latency risk on a long-lived server. Please push offset/length into the KVStoreView (skip/max) and cap the default length.

[sunchao]

[P3] There is also a concrete correctness case for moving this pagination into the KVStore view: offset + length is evaluated as an Int, so ?offset=1&length=2147483647 wraps the end index to Int.MinValue and returns an empty page even when rows remain. Using skip(offset).max(length) as suggested above avoids the overflow; please cover this boundary as well.

[sunchao]

Re-reviewed the current head after the latest fix. The original jobTag path-segment issue is fixed. I found two remaining P2s inline.

Non-blocking: the PR description still lists /operations/[job-tag] and four tests; the implementation now uses operationId and contains six tests.

[sunchao]

[P2] Keep the full operation identity in this lookup

operationId is only unique within an execution key: the manager keys operations by (userId, sessionId, operationId), and clients may supply any valid UUID. Two sessions can therefore legally create distinct executions with the same UUID. Both rows are returned by this secondary-index range, but headOption silently selects one, so following the ID from the other /operations list entry returns the wrong execution. Please include enough user/session identity in the route and lookup (or accept the unique jobTag as a query parameter), and add a duplicate-operation-ID-across-sessions test.

[sunchao]

[P2] Invalidate disk stores that lack this index

Existing disk-backed History Server stores were written without operationId index entries. Since AppStatusStore.CURRENT_VERSION is unchanged, loadDiskStore reopens them without replay, while LevelDB and RocksDB populate secondary indexes only on writes and do not backfill an added index on open. The operation list still reads the old natural records, but every detail lookup for those cached applications returns 404 until the store is rebuilt. Please bump the store version or add migration/fallback behavior, with an old-store reopen test.