From 3962ebbb370193acc0511bfff331666d96d3f5eb Mon Sep 17 00:00:00 2001
From: Gabor Roczei <1918366+roczei@users.noreply.github.com>
Date: Mon, 10 Aug 2026 16:49:34 +0200
Subject: [PATCH] [LIVY-1041] Add Spark 4 support and make it the default build
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## What changes were proposed in this pull request?

This PR adds support for building and running Apache Livy against
Apache Spark 4.x (pinned to 4.1.2). Spark 4 is now the default build
(`mvn package` with no extra profiles); Spark 3.x remains available via
`-Pspark3 -Pscala-2.12`.

Spark 4 introduced breaking changes that required widespread updates:
JDK 17 minimum, Scala 2.13 only, Netty 4.2.x, a rewritten
Scala REPL, and json4s 4.0.7.

**Build / POM changes**
- pom.xml: root properties default to Spark 4.1.2, Scala 2.13.17,
  Hadoop 3.4.1, and `java.version=17` so `mvn package` works without
  profile flags. The existing `spark3` and `scala-2.12` profiles
  override those defaults for legacy Spark 3 builds; dedicated
  `spark4` / `scala-2.13` profiles are no longer needed. Per Gyorgy's
  suggestion on [LIVY-1069](https://lists.apache.org/thread/xmb0hlqtmh0po2o75c025047x26lor9c),
  the latest supported Spark/Scala versions are now the default rather
  than requiring explicit `-P` flags.
- JDK handling: maven-enforcer follows the active profile — JDK 17 for the default
  Spark 4 build, JDK 8 minimum when `-Pspark3` sets `java.version=1.8`.
- maven-shade-plugin 3.5.0 -> 3.6.2 (bundles an ASM version that
  understands Java 22 bytecode, needed to shade Jackson 2.18.2's
  `META-INF/versions/22` multi-release classes).
- repl/pom.xml: added `json4s-jackson-core` to the shade includes
  (json4s 4.0.7 split `JsonMethods` into a separate artifact; missing
  it caused a NoClassDefFoundError in PythonInterpreter at runtime).
  Also pulls `livy-test-lib` in as a test dependency so the repl tests
  can use the new shared `ScalaVersionAware` trait.
- New Scala 2.13 module wrapper POMs: `core/scala-2.13`,
  `repl/scala-2.13`, `scala-api/scala-2.13`.
- README.md: documented default Spark 4 build and Spark 3 profile
  usage; Spark 4 Python compatibility — supported Python versions for
  Spark 4.1 are 3.10 – 3.14.

**Spark 4 / Scala 2.13 source-level fixes**
- `repl/scala-2.13/SparkInterpreter.scala`: new implementation ported
  to the Spark 4 SparkILoop (`shell.ILoop`, `PrintWriter`,
  `createInterpreter(settings)`, operations via `sparkILoop.intp`,
  `ReplCompletion`). To make extra JARs supplied via
  `spark.jars.packages` visible to `import ...` we feed those JARs to
  the Scala compiler through the `-classpath` argument at Settings
  construction time, instead of via a post-init
  `IMain.addUrlsToClassPath` call. In Scala 2.13 the latter does not
  reliably register URLs with the compiler's `platform.classPath`, so
  `import org.codehaus.plexus.util._` fails with `object plexus is not
  a member of package org.codehaus`. A new `collectUserJarsClasspath()`
  helper walks the context classloader chain to Spark's
  `MutableURLClassLoader`, filters out `livy-*` and the wrong-version
  `scala-reflect` (same rules as the old `addUrls` path), and joins the
  resulting file paths with `File.pathSeparator`. If the chain has no
  `MutableURLClassLoader` we log a warning and skip the extra
  `-classpath` entry rather than passing an empty argument.

- `AbstractSparkInterpreter.scala`:
  * `parseError` skips the leading caret line so `ename` lands on the
    "error: ..." message on both Scala 2.12 and 2.13 (2.13 prints the
    caret pointer before the message). Implemented with
    `lines.indexWhere(_.trim != "^") ... lines.patch(idx, Nil, 1)`.
  * New `isEffectivelyEmpty` helper strips block/line comments and
    returns success for comment-only inputs (Scala 2.12 accepted them
    silently; 2.13 rejects them as compile errors).
- `repl/Session.scala`: SparkR `setJobGroup` match now accepts `"4"`
  alongside `"2" | "3"`.
- `server/interactive/InteractiveSession.scala`: `datanucleusJars` and
  `mergeHiveSiteAndHiveDeps` now take an extra `scalaVersion`
  parameter, adds `case 3 | 4` to the major-version match and replaces
  the hard-coded `assembly/target/scala-2.12/jars` with
  `assembly/target/scala-$scalaVersion/jars`.
- `server/batch/BatchSession.scala`: pass `--verbose` to spark-submit
  under `LivyConf.TEST_MODE` so tests can grep the resolved arguments
  out of the child's log.
- `rsc/driver/SparkEntries.java`: new reflection-based
  `hiveClassesArePresent` helper that probes
  `org.apache.spark.sql.classic.SparkSession$` (Spark 4) first and
  falls back to `org.apache.spark.sql.SparkSession$` (Spark 3).
- `utils/SparkProcessBuilder.scala`: new `verbose(v: Boolean)` mutator
  and an internal `_verbose` flag that appends `--verbose` to
  `spark-submit` when set (used by `BatchSession`).
- `utils/LivySparkUtils.scala`: `(4, 0)` and `(4, 1)` -> `"2.13"`
  entries in `sparkScalaVersionMap`; `MAX_VERSION` bumped from
  `(3, 6)` to `(4, 2)`.
- `repl/SparkRInterpreter.scala`: log a warning on Spark 4 that
  SparkR was deprecated in SPARK-49347 and may be removed in a future
  Spark release.
- `scala-api/scalaapi/package.scala`: inline comment documenting that
  `Duration.isFinite` is a parameterless def in Scala 2.13 (took an
  empty parameter list in 2.12); dropping the parens compiles on both.
- Thriftserver Scala/json4s 4.x updates:
  * `session/Get{Columns,Functions,Schemas,Tables}Job.java`:
    `scala.collection.JavaConversions.seqAsJavaList` (removed in Scala
    2.13) replaced with
    `scala.collection.JavaConverters.seqAsJavaListConverter(...).asJava()`.
  * `types/DataTypeUtils.scala`: json4s 4.x `parse(input, useBigDecimal)`
    requires an `AsJsonInput[T]`; switch to the single-arg String
    overload `parse(sparkJson)`.
  * `LivyExecuteStatementOperation.scala`: explicit `res.toSeq` since
    Scala 2.13 no longer widens a mutable `Buffer` to `Seq` implicitly.
  * `ThriftServerSuites.scala`: `LIVY-571` assertion now also accepts
    the Spark 4 error string `[SCHEMA_NOT_FOUND] The schema
    `spark_catalog`.`invalid_database` cannot be found`.
- `InteractiveSessionServlet.scala`: `logs.asJava` -> `logs.toSeq.asJava`
  so Scala 2.13 picks the `Seq[A]` -> `java.util.List` `asJava`
  overload instead of erroring on ambiguity.

**Thriftserver libthrift pre-0.16 / 0.16 compatibility**
The Livy Thrift binary CLI service was written against libthrift 0.9.3
(the version pinned in the root pom.xml). Spark 3 keeps a compatible
pre-0.16 libthrift on the runtime classpath (e.g. 0.12.0 in Spark 3.3),
but Spark 4 pulls in libthrift 0.16.0 transitively through spark-hive,
and 0.16 removed four `TThreadPoolServer.Args` builder methods that
Livy relied on: `requestTimeout`, `requestTimeoutUnit`,
`beBackoffSlotLength` and `beBackoffSlotLengthUnit`. Compiling against
0.9.3 but running against 0.16.0 made the mini-cluster Thrift server
abort during startup with a `NoSuchMethodError` on
`TThreadPoolServer$Args.requestTimeout(int)`, which in turn hung
JdbcIT on the SASL handshake. `ThriftBinaryCLIService` now invokes
those four builders reflectively via a small `applyOptionalArg`
helper: they run unchanged on Spark 3 classpaths and are silently
skipped on Spark 4, where they no longer exist. The functional loss
on Spark 4 is limited to the login-timeout / backoff knobs, which
have no equivalent in the newer libthrift API.

**Test fix: Scala 2.13 REPL output shape changes in InteractiveIT**
The Scala 2.13 REPL prints results differently from Scala 2.12 in
four ways that `InteractiveIT` was too strict about; the default
Spark 4 build (Scala 2.13 only) trips all four. Fixed in place:
- Value results: `val res0: Int = 2` (2.13) vs. `res0: Int = 2`
  (2.12). Six `verifyResult` regexes (across `basic interactive
  session`, `user jars are properly imported ...` and `recover
  interactive session`) now accept an optional `val ` prefix.
- Compile errors: `error: not found: value abcde` (2.13) vs.
  `<console>:12: error: not found: value abcde` (2.12). The
  `<source>:<line>: ` location marker is now optional in the
  `verifyError` regex.
- Class definitions: `class Item` (2.13) vs. `defined class Item`
  (2.12). The `case class Item(i: Int)` check in `user jars are
  properly imported ...` now accepts an optional `defined ` prefix.
- Deprecation-warning header: Scala 2.13 prints a
  `warning: 1 deprecation ...` line above the value binding when the
  expression uses a deprecated API. `SparkContext.parallelize` is
  deprecated on Spark 4, so
  `val rdd = sc.parallelize(Array.fill(10){...})` prints the warning
  before the `val rdd: ...` line. The `verifyResult` regex now uses
  `(?s)(?:warning:.*\n)?(?:val )?rdd.*` to accept the warning prefix
  and the DOTALL semantics needed to span the newline.
Both forms match on the default Scala 2.13 build and `-Pscala-2.12`.

**Test fix: Spark 4 SQLContext / DataFrame runtime class in InteractiveIT**
Spark 4 moved the runtime `SQLContext` and `DataFrame` implementations
into `org.apache.spark.sql.classic`; only the compile-time aliases
remain at `org.apache.spark.sql.SQLContext` / `org.apache.spark.sql.DataFrame`.
The Scala 2.13 REPL surfaces the runtime class name, so Spark 4
prints `sql: org.apache.spark.sql.classic.SQLContext = ...` and
`val df: org.apache.spark.sql.classic.DataFrame = ...`, where Spark 3
prints the alias form. Two `InteractiveIT` regexes now allow an
optional `classic.` package qualifier so the same expectations match
both the default Spark 4 build and `-Pspark3`:
- `sql: org.apache.spark.sql.(?:classic\.)?SQLContext = ...`
- `(?:val )?df: org.apache.spark.sql.(?:classic\.)?DataFrame`
The old `Pattern.quote("df: org.apache.spark.sql.DataFrame")` literal
was too strict for Spark 4 and caused the `basic interactive session`
case to fail.

**Cross-version test helpers: `ScalaVersionAware` trait**
New `test-lib/.../ScalaVersionAware` trait exposes the small
Scala-2.12 vs 2.13 REPL-output differences as reusable fragments:
- `optionalValPrefix` (`"val "` on 2.13, `""` on 2.12) for exact-match
  expected strings.
- `optionalValPrefixRegex` / `optionalDefinedPrefixRegex` /
  `optionalWarningPrefixRegex` regex fragments for the corresponding
  `verifyResult` regex assertions.
Detection uses the runtime `scala.util.Properties.versionNumberString`
so the same test class works regardless of which Scala the artifact
was compiled against. Mixed into `BaseInterpreterSpec`,
`BaseSessionSpec` and `InteractiveIT`. Downstream specs
(`ScalaInterpreterSpec`, `SharedSessionSpec`, `SparkSessionSpec`,
`InteractiveIT`) rewrote their expected-value strings against these
fragments; some hard-`equal` assertions were relaxed to
`include`/`fullyMatch regex` where the 2.13 REPL adds a deprecation
banner ahead of the result or renumbers `resN` slots differently.

**Server/unit tests: cross-Scala tolerance in `InteractiveSessionSpec`**
The `should get scala version` case now decomposes the JSON result by
hand and asserts `text/plain` equals either `res0: Int = 3\n` or
`val res0: Int = 3\n`, plus separate `status` / `execution_count`
assertions. Straight equality with a decomposed Map broke on the
Spark 4 driver where the 2.13 REPL adds the `val ` prefix.

**Unit tests: `LivySparkUtilsSuite`**
Added `4.0.0` and `4.1.2` to both the supported-version list and the
`defaultSparkScalaVersion` -> `"2.13"` expectations.

**scala-api tests**
- `ScalaClientTestUtils.scala`: `context.sc.parallelize(buffer, ...)`
  -> `context.sc.parallelize(buffer.toSeq, ...)` (2.13 no longer
  widens a mutable `ArrayBuffer` to `Seq` implicitly for the
  `parallelize` signature).
- `ScalaJobHandleTest.scala`: `Duration.Undefined` -> `Duration.Inf`
  (2.13's `Await.ready` rejects `Undefined` with "Cannot wait for
  Undefined duration of time"); assertion switched from
  `verify(mockJobHandle, times(1)).get()` to
  `verify(mockJobHandle, atLeastOnce()).isDone`, because 2.13's
  `Await.ready` short-circuits when `isCompleted` is already true and
  does not call the underlying `ready(atMost)` -- so `.get()` is not
  observed on 2.13. `isDone` is exercised by both versions.

**Repl tests: `SparkInterpreterSpec` move**
`SparkInterpreterSpec.scala` moved from `repl/scala-2.12/src/test/...`
into the shared `repl/src/test/scala/...` tree (it was byte-identical
between scala-2.12 and scala-2.13). A new `should skip leading caret
lines in Scala 2.13 error format.` test case was added for the caret-
first `parseError` branch, so the shared spec runs three cases under
both the default Scala 2.13 build and `-Pscala-2.12`.

**CI matrix**
- `.github/workflows/unit-tests.yaml` and `integration-tests.yaml`
  converted the flat `maven_profile` x `jdk_path` matrix to an
  explicit `include:` list. The default spark4 build (no extra Maven
  profiles) runs on JDK 17; Spark 3 matrix entries use
  `-Pscala-2.12 -Pspark3`.

## How was this patch tested?

- Unit tests: `mvn verify -Pthriftserver` passes all 19 modules on JDK
  17 (macOS aarch64) with the default Spark 4 build. Verified modules
  include livy-rsc (40 tests), livy-repl_2.13 (87 tests), livy-server
  (205 tests), livy-client-http (17 tests), livy-scala-api_2.13
  (17 tests), and the thriftserver integration suite (27 tests:
  HttpThriftServerSuite + BinaryThriftServerSuite). The
  `SparkInterpreterSpec` move added one new caret-first `parseError`
  test case; the shared spec now has 3 cases and runs identically
  under both the default Scala 2.13 build and `-Pscala-2.12`.
- Integration tests (`mvn integration-test -Pspark3 -Pscala-2.12 -pl
  :livy-integration-test`) pass.
- GitHub Actions CI runs both spark3 and spark4 matrix entries for
  unit and integration test workflows.

## Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 4.7)
---
 .github/workflows/integration-tests.yaml      |  13 +-
 .github/workflows/unit-tests.yaml             |  13 +-
 README.md                                     |  26 ++-
 core/scala-2.13/pom.xml                       |  48 ++++++
 dev/docker/livy-dev-base/Dockerfile           |   2 +-
 .../org/apache/livy/test/InteractiveIT.scala  |  53 ++++--
 pom.xml                                       |  43 +++--
 repl/pom.xml                                  |  16 ++
 repl/scala-2.13/pom.xml                       |  34 ++++
 .../apache/livy/repl/SparkInterpreter.scala   | 163 ++++++++++++++++++
 .../livy/repl/AbstractSparkInterpreter.scala  |  35 +++-
 .../scala/org/apache/livy/repl/Session.scala  |   2 +-
 .../apache/livy/repl/SparkRInterpreter.scala  |   5 +
 .../livy/repl/BaseInterpreterSpec.scala       |   4 +-
 .../apache/livy/repl/BaseSessionSpec.scala    |   3 +-
 .../livy/repl/ScalaInterpreterSpec.scala      |  64 +++++--
 .../apache/livy/repl/SharedSessionSpec.scala  |  20 +--
 .../livy/repl/SparkInterpreterSpec.scala      |  15 ++
 .../apache/livy/repl/SparkSessionSpec.scala   |  28 +--
 .../apache/livy/rsc/driver/SparkEntries.java  |  34 +++-
 .../org/apache/livy/rsc/TestSparkClient.java  |   7 +-
 scala-api/scala-2.13/pom.xml                  |  31 ++++
 .../livy/scalaapi/ScalaClientTestUtils.scala  |   2 +-
 .../livy/scalaapi/ScalaJobHandleTest.scala    |  10 +-
 .../livy/server/batch/BatchSession.scala      |   5 +
 .../interactive/InteractiveSession.scala      |  19 +-
 .../InteractiveSessionServlet.scala           |   6 +-
 .../apache/livy/utils/LivySparkUtils.scala    |   6 +-
 .../livy/utils/SparkProcessBuilder.scala      |  13 ++
 .../interactive/InteractiveSessionSpec.scala  |  13 +-
 .../livy/utils/LivySparkUtilsSuite.scala      |   4 +
 .../apache/livy/test/ScalaVersionAware.scala  |  50 ++++++
 .../LivyExecuteStatementOperation.scala       |   3 +-
 .../cli/ThriftBinaryCLIService.scala          |  36 +++-
 .../thriftserver/types/DataTypeUtils.scala    |   7 +-
 .../thriftserver/ThriftServerSuites.scala     |   9 +-
 .../thriftserver/session/GetColumnsJob.java   |   9 +-
 .../thriftserver/session/GetFunctionsJob.java |   7 +-
 .../thriftserver/session/GetSchemasJob.java   |   5 +-
 .../thriftserver/session/GetTablesJob.java    |   7 +-
 40 files changed, 729 insertions(+), 141 deletions(-)
 create mode 100644 core/scala-2.13/pom.xml
 create mode 100644 repl/scala-2.13/pom.xml
 create mode 100644 repl/scala-2.13/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala
 rename repl/{scala-2.12 => }/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala (78%)
 create mode 100644 scala-api/scala-2.13/pom.xml
 create mode 100644 test-lib/src/main/scala/org/apache/livy/test/ScalaVersionAware.scala

diff --git a/.github/workflows/integration-tests.yaml b/.github/workflows/integration-tests.yaml
index de362a5b8..1f8dd453f 100644
--- a/.github/workflows/integration-tests.yaml
+++ b/.github/workflows/integration-tests.yaml
@@ -28,11 +28,14 @@ jobs:
     container: ghcr.io/${{ github.repository_owner }}/livy-ci:latest
     strategy:
       matrix:
-        maven_profile:
-        - "-Pscala-2.12 -Pspark3"
-        jdk_path:
-        - "/usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java"
-        - "/usr/lib/jvm/java-17-openjdk-amd64/bin/java"
+        include:
+          - maven_profile: "-Pscala-2.12 -Pspark3"
+            jdk_path: "/usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java"
+          - maven_profile: "-Pscala-2.12 -Pspark3"
+            jdk_path: "/usr/lib/jvm/java-17-openjdk-amd64/bin/java"
+          # spark4 is the default build; JDK 17+ required.
+          - maven_profile: ""
+            jdk_path: "/usr/lib/jvm/java-17-openjdk-amd64/bin/java"
     steps:
     -
       name: Checkout
diff --git a/.github/workflows/unit-tests.yaml b/.github/workflows/unit-tests.yaml
index b3dd9a2e5..c05961af3 100644
--- a/.github/workflows/unit-tests.yaml
+++ b/.github/workflows/unit-tests.yaml
@@ -29,11 +29,14 @@ jobs:
     container: ghcr.io/${{ github.repository_owner }}/livy-ci:latest
     strategy:
       matrix:
-        maven_profile:
-        - "-Pscala-2.12 -Pspark3"
-        jdk_path:
-        - "/usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java"
-        - "/usr/lib/jvm/java-17-openjdk-amd64/bin/java"
+        include:
+          - maven_profile: "-Pscala-2.12 -Pspark3"
+            jdk_path: "/usr/lib/jvm/java-8-openjdk-amd64/jre/bin/java"
+          - maven_profile: "-Pscala-2.12 -Pspark3"
+            jdk_path: "/usr/lib/jvm/java-17-openjdk-amd64/bin/java"
+          # spark4 is the default build; JDK 17+ required.
+          - maven_profile: ""
+            jdk_path: "/usr/lib/jvm/java-17-openjdk-amd64/bin/java"
     steps:
     -
       name: Checkout
diff --git a/README.md b/README.md
index 32ee32d98..305555857 100644
--- a/README.md
+++ b/README.md
@@ -78,12 +78,12 @@ You can also use the provided [Dockerfile](./dev/docker/livy-dev-base/Dockerfile
 git clone https://github.com/apache/livy.git
 cd livy
 docker build -t livy-ci dev/docker/livy-dev-base/
-docker run --rm -it -v $(pwd):/workspace -v $HOME/.m2:/root/.m2 livy-ci mvn package -Pspark3 -Pscala-2.12
+docker run --rm -it -v $(pwd):/workspace -v $HOME/.m2:/root/.m2 livy-ci mvn package
 ```
 
 > **Note**: The `docker run` command maps the maven repository to your host machine's maven cache so subsequent runs will not need to download dependencies.
 
-By default Livy is built against Apache Spark 3.3.4, but the version of Spark used when running
+By default Livy is built against Apache Spark 4.1.2 with Scala 2.13, but the version of Spark used when running
 Livy does not need to match the version used to build Livy. Livy internally handles the differences
 between different Spark versions.
 
@@ -92,9 +92,19 @@ version of Spark without needing to rebuild.
 
 ### Build Profiles
 
-| Flag           | Purpose                                    |
-|----------------|--------------------------------------------|
-| -Phadoop2      | Choose Hadoop2 based build dependencies    |
-| -Pthriftserver | Build and test Livy Thrift Server modules  |
-| -Pspark3       | Choose Spark 3.x based build dependencies  |
-| -Pscala-2.12   | Choose Scala 2.12 based build dependencies |
+| Flag           | Purpose                                                                            |
+|----------------|------------------------------------------------------------------------------------|
+| -Phadoop2      | Choose Hadoop2 based build dependencies                                            |
+| -Pthriftserver | Build and test Livy Thrift Server modules                                          |
+| -Pspark3       | Choose Spark 3.x based build dependencies (use with `-Pscala-2.12`)                |
+| -Pscala-2.12   | Choose Scala 2.12 based build dependencies (use with `-Pspark3`)                   |
+
+Example — build against Spark 3:
+
+```
+mvn package -Pspark3 -Pscala-2.12
+```
+
+> **Note**: The default build targets Spark 4.1.2 and requires JDK 17 or JDK 21 with Scala 2.13 and Hadoop 3.4.1.
+> JDK 8, JDK 11 and Scala 2.12 are not supported by Spark 4. Supported Python
+> versions for Spark 4.1 are 3.10 – 3.14.
diff --git a/core/scala-2.13/pom.xml b/core/scala-2.13/pom.xml
new file mode 100644
index 000000000..541e1aedf
--- /dev/null
+++ b/core/scala-2.13/pom.xml
@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>org.apache.livy</groupId>
+  <artifactId>livy-core_2.13</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <parent>
+    <groupId>org.apache.livy</groupId>
+    <artifactId>livy-core-parent</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-jar-plugin</artifactId>
+        <executions>
+          <execution>
+            <goals>
+              <goal>test-jar</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
\ No newline at end of file
diff --git a/dev/docker/livy-dev-base/Dockerfile b/dev/docker/livy-dev-base/Dockerfile
index eae8e7943..a0d210a60 100644
--- a/dev/docker/livy-dev-base/Dockerfile
+++ b/dev/docker/livy-dev-base/Dockerfile
@@ -72,7 +72,7 @@ ENV PATH="$HOME/pyenv/shims:$HOME/pyenv/bin:$HOME/bin:$PATH"
 
 # Python 3.11 is chosen because it is in the officially supported range of
 # BOTH matrix profiles, per each Spark release's python/setup.py:
-#   * -Pspark4 (Spark 4.1.2): python_requires=">=3.10", classifiers list
+#   * default / Spark 4.1.2: python_requires=">=3.10", classifiers list
 #     3.10 / 3.11 / 3.12 / 3.13 / 3.14.
 #   * -Pspark3 (Spark 3.5.6): python_requires=">=3.8", classifiers list
 #     3.8 / 3.9 / 3.10 / 3.11.
diff --git a/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala b/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
index ec170a2ad..08cb646b1 100644
--- a/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
+++ b/integration-test/src/test/scala/org/apache/livy/test/InteractiveIT.scala
@@ -30,27 +30,46 @@ import org.apache.livy.rsc.RSCConf
 import org.apache.livy.sessions._
 import org.apache.livy.test.framework.{BaseIntegrationTestSuite, LivyRestClient}
 
-class InteractiveIT extends BaseIntegrationTestSuite {
+class InteractiveIT extends BaseIntegrationTestSuite with ScalaVersionAware {
   test("basic interactive session") {
     withNewSession(Spark) { s =>
       s.run("val sparkVersion = sc.version").result().left.foreach(info(_))
       s.run("val scalaVersion = util.Properties.versionString").result().left.foreach(info(_))
-      s.run("1+1").verifyResult("res0: Int = 2\n")
+      // Scala 2.13's REPL prints a `val ` prefix before result names
+      // (`val res0: Int = 2`) whereas Scala 2.12 prints just `res0: Int = 2`.
+      // Accept both forms so the test passes on either the default Scala 2.13
+      // build or -Pscala-2.12.
+      s.run("1+1").verifyResult(s"${optionalValPrefixRegex}res0: Int = 2\n")
 
       // Ignore the following line if running on a external cluster due to config differences
       // with the mini cluster
-      s.run("""sc.getConf.get("spark.executor.instances")""").verifyResult("res1: String = 1\n")
+      s.run("""sc.getConf.get("spark.executor.instances")""")
+        .verifyResult(s"${optionalValPrefixRegex}res1: String = 1\n")
 
+      // Spark 4 relocated SQLContext into the `org.apache.spark.sql.classic`
+      // package; Spark 3 keeps it directly under `org.apache.spark.sql`. Match
+      // either shape (`SQLContext` or `classic.SQLContext`) and tolerate the
+      // Scala 2.13 REPL's `val ` prefix as elsewhere in this test.
       s.run("val sql = spark.sqlContext").verifyResult(
-        ".*" + Pattern.quote(
-        "sql: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext") + ".*")
-      s.run("abcde").verifyError(evalue = ".*?:[0-9]+: error: not found: value abcde.*")
+        ".*sql: org\\.apache\\.spark\\.sql\\.(?:classic\\.)?SQLContext = " +
+        "org\\.apache\\.spark\\.sql\\.(?:classic\\.)?SQLContext.*")
+      // Scala 2.12's REPL prefixes compile errors with a "<source>:<line>:"
+      // location marker (e.g. "<console>:12: error: not found: value abcde"),
+      // while Scala 2.13 drops the marker and just emits "error: not found:
+      // value abcde". Accept both.
+      s.run("abcde").verifyError(evalue = ".*(?:.*?:[0-9]+: )?error: not found: value abcde.*")
       s.run("throw new IllegalStateException()")
         .verifyError(evalue = ".*java\\.lang\\.IllegalStateException.*")
 
-      // Verify query submission
+      // Verify query submission. Spark 4's Scala 2.13 REPL surfaces the runtime
+      // class name (`org.apache.spark.sql.classic.DataFrame`) rather than the
+      // compile-time alias (`org.apache.spark.sql.DataFrame`) that Spark 3
+      // prints, and prefixes the identifier with `val ` like elsewhere in this
+      // test. Accept both shapes.
       s.run(s"""val df = spark.createDataFrame(Seq(("jerry", 20), ("michael", 21)))""")
-        .verifyResult(".*" + Pattern.quote("df: org.apache.spark.sql.DataFrame") + ".*")
+        .verifyResult(
+          s".*${optionalValPrefixRegex}df: " +
+            "org\\.apache\\.spark\\.sql\\.(?:classic\\.)?DataFrame.*")
       s.run("df.createOrReplaceTempView(\"people\")").result()
       s.run("SELECT * FROM people", Some(SQL)).verifyResult(".*\"jerry\",20.*\"michael\",21.*")
 
@@ -167,9 +186,15 @@ class InteractiveIT extends BaseIntegrationTestSuite {
       s.run("import org.codehaus.plexus.util._").verifyResult("import org.codehaus.plexus.util._\n")
 
       // Check does SparkContext see classes defined by Scala interpreter.
-      s.run("case class Item(i: Int)").verifyResult("defined class Item\n")
+      // Scala 2.12's REPL reports `defined class Item`; Scala 2.13 emits the
+      // shorter `class Item`. Accept both.
+      s.run("case class Item(i: Int)").verifyResult(s"${optionalDefinedPrefixRegex}class Item\n")
+      // Scala 2.13's REPL prefixes with `val ` (e.g. `val rdd: ...`); 2.12
+      // omits it. Spark 4 (Scala 2.13) also prints a deprecation warning
+      // before the value binding because `parallelize` is now deprecated on
+      // SparkContext -- accept an optional warning header.
       s.run("val rdd = sc.parallelize(Array.fill(10){new Item(scala.util.Random.nextInt(1000))})")
-        .verifyResult("rdd.*")
+        .verifyResult(s"(?s)${optionalWarningPrefixRegex}${optionalValPrefixRegex}rdd.*")
       s.run("rdd.count()").verifyResult(".*= 10\n")
     }
   }
@@ -188,15 +213,17 @@ class InteractiveIT extends BaseIntegrationTestSuite {
   test("recover interactive session") {
     withNewSession(Spark) { s =>
       val stmt1 = s.run("1")
-      stmt1.verifyResult("res0: Int = 1\n")
+      // Scala 2.13's REPL renders value results as `val res0: Int = 1`
+      // whereas Scala 2.12 prints `res0: Int = 1`; accept both.
+      stmt1.verifyResult(s"${optionalValPrefixRegex}res0: Int = 1\n")
 
       restartLivy()
 
       // Verify session still exists.
       s.verifySessionIdle()
-      s.run("2").verifyResult("res1: Int = 2\n")
+      s.run("2").verifyResult(s"${optionalValPrefixRegex}res1: Int = 2\n")
       // Verify statement result is preserved.
-      stmt1.verifyResult("res0: Int = 1\n")
+      stmt1.verifyResult(s"${optionalValPrefixRegex}res0: Int = 1\n")
 
       s.stop()
 
diff --git a/pom.xml b/pom.xml
index 738c166ae..4bea4befd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -81,16 +81,15 @@
     <asynchttpclient.version>2.10.1</asynchttpclient.version>
     <hadoop.scope>compile</hadoop.scope>
     <slf4j.version>1.7.36</slf4j.version>
-    <spark.scala-2.12.version>3.3.4</spark.scala-2.12.version>
-    <spark.version>${spark.scala-2.12.version}</spark.version>
+    <spark.version>4.1.2</spark.version>
     <kubernetes.client.version>5.6.0</kubernetes.client.version>
     <hive.version>3.0.0</hive.version>
     <commons-codec.version>1.15</commons-codec.version>
     <commons-lang3.version>3.17.0</commons-lang3.version>
     <httpclient.version>4.5.14</httpclient.version>
     <httpcore.version>4.4.16</httpcore.version>
-    <jackson.version>2.12.7</jackson.version>
-    <jackson-databind.version>2.12.7.1</jackson-databind.version>
+    <jackson.version>2.18.2</jackson.version>
+    <jackson-databind.version>2.18.2</jackson-databind.version>
     <jacoco.version>0.8.13</jacoco.version>
     <javax.servlet-api.version>3.1.0</javax.servlet-api.version>
     <jetty.version>9.4.56.v20240826</jetty.version>
@@ -103,7 +102,10 @@
          HealthCheckFilter at runtime (the class moved in Dropwizard 4.x). -->
     <metrics.version>4.2.19</metrics.version>
     <mockito.version>1.10.19</mockito.version>
-    <netty.version>4.1.86.Final</netty.version>
+    <!-- Spark 4.1.2 depends on Netty 4.2.7 and calls the new 4.2-only
+         `KQueueIoHandler`; pinning to a 4.1.x line here would let Livy's
+         `netty-all` win and hide that class from Spark's classpath. -->
+    <netty.version>4.2.7.Final</netty.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <!-- Scalatest 3.2.9 is required by Scalatra 2.8.x (needed for Scala
          2.13 support). All test suites are written against the 3.2 API,
@@ -113,7 +115,6 @@
          scalatest (e.g. 3.2.9.0 for scalatest 3.2.9). -->
     <scalatestplus.mockito.version>3.2.9.0</scalatestplus.mockito.version>
     <scalatra.version>2.8.4</scalatra.version>
-    <java.version>1.8</java.version>
     <owasp.version>12.1.9</owasp.version>
     <extraJavaTestArgs>
       -XX:+IgnoreUnrecognizedVMOptions
@@ -133,15 +134,19 @@
     <test.redirectToFile>true</test.redirectToFile>
     <execution.root>${user.dir}</execution.root>
     <spark.home>${execution.root}/dev/spark</spark.home>
-    <!-- profile controlled parameters, we are setting the defaults here so we can run "mvn package" with default profile values -->
-    <!-- hadoop2 -->
-    <hadoop.major-minor.version>2</hadoop.major-minor.version>
-    <hadoop.version>2.7.3</hadoop.version>
-    <scala.binary.version>2.12</scala.binary.version>
-    <scala.version>2.12.18</scala.version>
-    <java.version>1.8</java.version>
-    <py4j.version>0.10.9</py4j.version>
-    <json4s.version>3.5.3</json4s.version>
+    <!-- Profile-controlled parameters: defaults target Spark 4 + Scala 2.13 so
+         `mvn package` works without extra flags. Use -Pspark3 -Pscala-2.12 for
+         Spark 3 builds. -->
+    <hadoop.major-minor.version>3</hadoop.major-minor.version>
+    <hadoop.version>3.4.1</hadoop.version>
+    <scala.binary.version>2.13</scala.binary.version>
+    <!-- Match the Scala patch Spark 4.1.2 was built against. Spark's
+         Catalyst calls `MurmurHash3.caseClassHash$default$2()` which was
+         added in 2.13.17; older 2.13.x runs fail with NoSuchMethodError. -->
+    <scala.version>2.13.17</scala.version>
+    <java.version>17</java.version>
+    <py4j.version>0.10.9.9</py4j.version>
+    <json4s.version>4.0.7</json4s.version>
     <spark.bin.name>spark-${spark.version}-bin-hadoop${hadoop.major-minor.version}</spark.bin.name>
     <spark.bin.download.url>
       https://archive.apache.org/dist/spark/spark-${spark.version}/${spark.bin.name}.tgz
@@ -1071,7 +1076,11 @@
         <plugin>
           <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-shade-plugin</artifactId>
-          <version>3.5.0</version>
+          <!-- 3.6.2 bundles an ASM version that understands Java 22 bytecode
+               (major-version 66). Jackson 2.18.2 ships JDK 22 multi-release
+               classes under META-INF/versions/22, which 3.5.0's older ASM
+               refused to shade with `Unsupported class file major version`. -->
+          <version>3.6.2</version>
         </plugin>
 
         <plugin>
@@ -1425,6 +1434,7 @@
       </properties>
     </profile>
     <profile>
+      <!-- Use with `-Pspark3` for Spark 3 builds. -->
       <id>scala-2.12</id>
       <properties>
         <scala.binary.version>2.12</scala.binary.version>
@@ -1442,6 +1452,7 @@
     </profile>
 
     <profile>
+      <!-- Use with `-Pscala-2.12` for Spark 3 builds. -->
       <id>spark3</id>
       <properties>
         <spark.version>3.5.6</spark.version>
diff --git a/repl/pom.xml b/repl/pom.xml
index 2c950a2b2..cd7f0ab0d 100644
--- a/repl/pom.xml
+++ b/repl/pom.xml
@@ -60,6 +60,13 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>livy-test-lib</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>
       <artifactId>jackson-core</artifactId>
@@ -205,6 +212,15 @@
                   <include>org.json4s:json4s-ast_${scala.binary.version}</include>
                   <include>org.json4s:json4s-core_${scala.binary.version}</include>
                   <include>org.json4s:json4s-jackson_${scala.binary.version}</include>
+                  <!-- json4s 4.x moved `JsonMethods` into a separate
+                       `json4s-jackson-core` module; without shading it too,
+                       the shaded jar is missing `JsonMethods.class` and
+                       PythonInterpreter fails at runtime with
+                       NoClassDefFoundError. The include is scoped to the
+                       artifactId prefix so builds against older json4s
+                       versions (which don't publish this module) still
+                       succeed. -->
+                  <include>org.json4s:json4s-jackson-core_${scala.binary.version}</include>
                   <include>org.json4s:json4s-scalap_${scala.binary.version}</include>
                   <include>com.esotericsoftware:kryo-shaded</include>
                 </includes>
diff --git a/repl/scala-2.13/pom.xml b/repl/scala-2.13/pom.xml
new file mode 100644
index 000000000..f63f0972e
--- /dev/null
+++ b/repl/scala-2.13/pom.xml
@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>org.apache.livy</groupId>
+  <artifactId>livy-repl_2.13</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <parent>
+    <groupId>org.apache.livy</groupId>
+    <artifactId>livy-repl-parent</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+</project>
\ No newline at end of file
diff --git a/repl/scala-2.13/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala b/repl/scala-2.13/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala
new file mode 100644
index 000000000..4339c19fa
--- /dev/null
+++ b/repl/scala-2.13/src/main/scala/org/apache/livy/repl/SparkInterpreter.scala
@@ -0,0 +1,163 @@
+/*
+ * 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.livy.repl
+
+import java.io.{File, PrintWriter}
+import java.net.{URL, URLClassLoader}
+import java.nio.file.{Files, Paths}
+
+import scala.tools.nsc.Settings
+import scala.tools.nsc.interpreter.{IMain, Repl}
+import scala.tools.nsc.interpreter.Results.Result
+import scala.tools.nsc.interpreter.shell.{Completion, NoCompletion, ReplCompletion}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.repl.SparkILoop
+
+/**
+ * Spark 4.x / Scala 2.13 implementation of the Spark interpreter
+ */
+class SparkInterpreter(protected override val conf: SparkConf) extends AbstractSparkInterpreter {
+
+  private var sparkILoop: SparkILoop = _
+
+  override def start(): Unit = {
+    require(sparkILoop == null)
+
+    val rootDir = conf.get("spark.repl.classdir", System.getProperty("java.io.tmpdir"))
+    val outputDir = Files.createTempDirectory(Paths.get(rootDir), "spark").toFile
+    outputDir.deleteOnExit()
+    conf.set("spark.repl.class.outputDir", outputDir.getAbsolutePath)
+
+    // Collect Spark's user JARs (from `spark.jars.packages`, `--jars`, etc.)
+    // and feed them to the Scala compiler through `-classpath` at construction
+    // time.
+    val userJarsClasspath = collectUserJarsClasspath()
+
+    val settings = new Settings()
+    val baseArgs = List(
+      "-Yrepl-class-based",
+      "-Yrepl-outdir", s"${outputDir.getAbsolutePath}")
+    val cpArgs = if (userJarsClasspath.nonEmpty) {
+      List("-classpath", userJarsClasspath)
+    } else {
+      Nil
+    }
+    settings.processArguments(baseArgs ++ cpArgs, true)
+    settings.usejavacp.value = true
+    settings.embeddedDefaults(Thread.currentThread().getContextClassLoader())
+
+    // Spark 4's SparkILoop takes (BufferedReader, PrintWriter)
+    sparkILoop = new SparkILoop(null, new PrintWriter(outputStream, true))
+    sparkILoop.createInterpreter(settings)
+
+    restoreContextClassLoader {
+      postStart()
+    }
+  }
+
+  /**
+   * Return Spark's `MutableURLClassLoader` URLs joined with
+   * `File.pathSeparator`, ready to pass to Scala compiler `-classpath`.
+   * Skips livy-* and scala-reflect jars; returns "" if no such classloader
+   * is found on the context classloader chain.
+   */
+  private def collectUserJarsClasspath(): String = {
+    var classLoader = Thread.currentThread().getContextClassLoader
+    while (classLoader != null &&
+      classLoader.getClass.getCanonicalName !=
+        "org.apache.spark.util.MutableURLClassLoader") {
+      classLoader = classLoader.getParent
+    }
+    if (classLoader == null) {
+      warn("Could not locate Spark's MutableURLClassLoader on the context " +
+        "classloader chain; user JARs from `spark.jars.packages` may not " +
+        "be visible to `import ...` inside the Scala interpreter.")
+      ""
+    } else {
+      val extraJarPath = classLoader.asInstanceOf[URLClassLoader].getURLs()
+        // Only real files -- getURLs() may include stale entries.
+        .filter { u => u.getProtocol == "file" && new File(u.getPath).isFile }
+        // Drop livy-* (would collide with the shaded repl classpath) and
+        // wrong scala-reflect version jars that some Spark packages depend on.
+        .filterNot { u =>
+          val name = Paths.get(u.toURI).getFileName.toString
+          name.startsWith("livy-") || name.contains("org.scala-lang_scala-reflect")
+        }
+      extraJarPath.foreach { p => debug(s"Adding $p to Scala interpreter's class path...") }
+      extraJarPath.map { u => new File(u.toURI).getAbsolutePath }
+        .mkString(File.pathSeparator)
+    }
+  }
+
+  override def close(): Unit = synchronized {
+    super.close()
+
+    if (sparkILoop != null) {
+      sparkILoop.closeInterpreter()
+      sparkILoop = null
+    }
+  }
+
+  override def addJar(jar: String): Unit = {
+    // Guard against the `_runtimeClassLoader == null` NPE inside
+    // `addUrlsToClassPath` (`urls.foreach(_runtimeClassLoader.addURL)`) on a
+    // fresh session that hasn't run any code yet. Calling `.classLoader` on
+    // the `Repl` interface triggers `ensureClassLoader()` -> `makeClassLoader()`
+    // in IMain, which initialises BOTH `_classLoader` (the
+    // AbstractFileClassLoader returned to us) AND `_runtimeClassLoader` (the
+    // URLClassLoader used by `addUrlsToClassPath`). We discard the returned
+    // value; we only need the side effect on `_runtimeClassLoader`.
+    sparkILoop.intp.classLoader
+    sparkILoop.intp.addUrlsToClassPath(new URL(jar))
+  }
+
+  override protected def isStarted(): Boolean = {
+    sparkILoop != null
+  }
+
+  override protected def interpret(code: String): Result = {
+    sparkILoop.intp.interpret(code)
+  }
+
+  override protected def completeCandidates(code: String, cursor: Int) : Array[String] = {
+    // Scala 2.13 replaced `PresentationCompilerCompleter` with
+    // `shell.ReplCompletion`, which takes a `Repl` (the interface `IMain`
+    // now implements). Instantiate it directly rather than by reflection.
+    val completer: Completion =
+      try new ReplCompletion(sparkILoop.intp.asInstanceOf[Repl])
+      catch { case _: Throwable => NoCompletion }
+    completer.complete(code, cursor, filter = false).candidates.map(_.name).toArray
+  }
+
+  override protected def valueOfTerm(name: String): Option[Any] = {
+    // IMain#valueOfTerm always returns None; read `$result` off the last request instead.
+    Option(sparkILoop.intp.asInstanceOf[IMain].lastRequest.lineRep.call("$result"))
+  }
+
+  override protected def bind(name: String,
+      tpe: String,
+      value: Object,
+      modifier: List[String]): Unit = {
+    // 2.12's `SparkILoop.beQuietDuring` moved to `intp.beQuietDuring` in 2.13;
+    // call the underlying `reporter.withoutPrintingResults` directly.
+    sparkILoop.intp.reporter.withoutPrintingResults {
+      sparkILoop.intp.bind(name, tpe, value, modifier)
+    }
+  }
+}
diff --git a/repl/src/main/scala/org/apache/livy/repl/AbstractSparkInterpreter.scala b/repl/src/main/scala/org/apache/livy/repl/AbstractSparkInterpreter.scala
index 0decf095d..64fbe0a9e 100644
--- a/repl/src/main/scala/org/apache/livy/repl/AbstractSparkInterpreter.scala
+++ b/repl/src/main/scala/org/apache/livy/repl/AbstractSparkInterpreter.scala
@@ -34,6 +34,23 @@ import org.apache.livy.rsc.driver.SparkEntries
 object AbstractSparkInterpreter {
   private[repl] val KEEP_NEWLINE_REGEX = """(?<=\n)""".r
   private val MAGIC_REGEX = "^%(\\w+)\\W*(.*)".r
+
+  /**
+   * True when `code` contains only whitespace and Scala comments (line or
+   * block), and nothing the Scala compiler would treat as a statement.
+   *
+   * The Scala 2.13 REPL rejects a source made up entirely of comments as a
+   * compile error, whereas 2.12 accepted it silently. Livy short-circuits
+   * these inputs so that behaviour matches across both Scala versions.
+   */
+  private[repl] def isEffectivelyEmpty(code: String): Boolean = {
+    // Strip block comments ((?s) DOTALL so `.` matches newlines across
+    // /* ... */) then line comments, and check whether anything meaningful
+    // remains.
+    val noBlock = code.replaceAll("(?s)/\\*.*?\\*/", "")
+    val noLine = noBlock.replaceAll("//[^\\n]*", "")
+    noLine.trim.isEmpty
+  }
 }
 
 abstract class AbstractSparkInterpreter extends Interpreter with Logging {
@@ -300,6 +317,11 @@ abstract class AbstractSparkInterpreter extends Interpreter with Logging {
     code match {
       case MAGIC_REGEX(magic, rest) =>
         executeMagic(magic, rest)
+      case _ if AbstractSparkInterpreter.isEffectivelyEmpty(code) =>
+        // Scala 2.13's REPL rejects a source with only comments as a compile
+        // error (2.12 quietly returned Success). Short-circuit here so both
+        // versions produce an empty-successful response.
+        Interpreter.ExecuteSuccess(TEXT_PLAIN -> "")
       case _ =>
         scala.Console.withOut(outputStream) {
           interpret(code) match {
@@ -327,13 +349,14 @@ abstract class AbstractSparkInterpreter extends Interpreter with Logging {
     //   at .error(<console>:11)
     //   ... 32 elided
 
-    // Return the first line as ename. Lines following as traceback.
-
     val lines = KEEP_NEWLINE_REGEX.split(stdout)
-    val ename = lines.headOption.map(_.trim).getOrElse("unknown error")
-    val traceback = lines.tail
-
-    (ename, traceback)
+    // Skip 2.13's leading caret line; falls through to head on 2.12 (message first).
+    val enameIdx = lines.indexWhere(l => l.trim.nonEmpty && l.trim != "^")
+    if (enameIdx < 0) {
+      (lines.headOption.map(_.trim).getOrElse("unknown error"), lines.tail.toSeq)
+    } else {
+      (lines(enameIdx).trim, lines.patch(enameIdx, Nil, 1).toSeq)
+    }
   }
 
   protected def restoreContextClassLoader[T](fn: => T): T = {
diff --git a/repl/src/main/scala/org/apache/livy/repl/Session.scala b/repl/src/main/scala/org/apache/livy/repl/Session.scala
index c1267bc45..acf7b4de2 100644
--- a/repl/src/main/scala/org/apache/livy/repl/Session.scala
+++ b/repl/src/main/scala/org/apache/livy/repl/Session.scala
@@ -348,7 +348,7 @@ class Session(
           case "1" =>
             (s"""setJobGroup(sc, "$jobGroup", "Job group for statement $jobGroup", FALSE)""",
              SparkR)
-          case "2" | "3" =>
+          case "2" | "3" | "4" =>
             (s"""setJobGroup("$jobGroup", "Job group for statement $jobGroup", FALSE)""", SparkR)
           case v =>
             throw new IllegalArgumentException(s"Unknown Spark major version [$v]")
diff --git a/repl/src/main/scala/org/apache/livy/repl/SparkRInterpreter.scala b/repl/src/main/scala/org/apache/livy/repl/SparkRInterpreter.scala
index 407762623..47c3f93b6 100644
--- a/repl/src/main/scala/org/apache/livy/repl/SparkRInterpreter.scala
+++ b/repl/src/main/scala/org/apache/livy/repl/SparkRInterpreter.scala
@@ -183,6 +183,11 @@ class SparkRInterpreter(
   override def kind: String = "sparkr"
   private[this] val isStarted = new CountDownLatch(1)
 
+  if (sparkMajorVersion >= 4) {
+    warn("SparkR is deprecated in Spark 4 (SPARK-49347); the SparkR interpreter " +
+      "may be removed in a future Spark release.")
+  }
+
   final override protected def waitUntilReady(): Unit = {
     // Set the option to catch and ignore errors instead of halting.
     sendRequest("options(error = dump.frames)")
diff --git a/repl/src/test/scala/org/apache/livy/repl/BaseInterpreterSpec.scala b/repl/src/test/scala/org/apache/livy/repl/BaseInterpreterSpec.scala
index a4345fa9f..2ac4cf3f6 100644
--- a/repl/src/test/scala/org/apache/livy/repl/BaseInterpreterSpec.scala
+++ b/repl/src/test/scala/org/apache/livy/repl/BaseInterpreterSpec.scala
@@ -21,8 +21,10 @@ import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 import org.apache.livy.LivyBaseUnitTestSuite
+import org.apache.livy.test.ScalaVersionAware
 
-abstract class BaseInterpreterSpec extends AnyFlatSpec with Matchers with LivyBaseUnitTestSuite {
+abstract class BaseInterpreterSpec
+    extends AnyFlatSpec with Matchers with LivyBaseUnitTestSuite with ScalaVersionAware {
 
   def createInterpreter(): Interpreter
 
diff --git a/repl/src/test/scala/org/apache/livy/repl/BaseSessionSpec.scala b/repl/src/test/scala/org/apache/livy/repl/BaseSessionSpec.scala
index 4d052006f..e8a80e372 100644
--- a/repl/src/test/scala/org/apache/livy/repl/BaseSessionSpec.scala
+++ b/repl/src/test/scala/org/apache/livy/repl/BaseSessionSpec.scala
@@ -35,9 +35,10 @@ import org.apache.livy.client.common.TestUtils
 import org.apache.livy.rsc.RSCConf
 import org.apache.livy.rsc.driver.{Statement, StatementState}
 import org.apache.livy.sessions._
+import org.apache.livy.test.ScalaVersionAware
 
 abstract class BaseSessionSpec(kind: Kind)
-    extends AnyFlatSpec with Matchers with LivyBaseUnitTestSuite {
+    extends AnyFlatSpec with Matchers with LivyBaseUnitTestSuite with ScalaVersionAware {
 
   implicit val formats = DefaultFormats
 
diff --git a/repl/src/test/scala/org/apache/livy/repl/ScalaInterpreterSpec.scala b/repl/src/test/scala/org/apache/livy/repl/ScalaInterpreterSpec.scala
index a13efeba1..25b62668e 100644
--- a/repl/src/test/scala/org/apache/livy/repl/ScalaInterpreterSpec.scala
+++ b/repl/src/test/scala/org/apache/livy/repl/ScalaInterpreterSpec.scala
@@ -27,30 +27,34 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
 
   implicit val formats = DefaultFormats
 
+  // `optionalValPrefix` (Scala-2.13's `val ` before bound-variable output, empty on
+  // 2.12) is provided by `BaseInterpreterSpec` so the same source works
+  // against both scala-2.12 and scala-2.13 builds.
+
   override def createInterpreter(): Interpreter =
     new SparkInterpreter(new SparkConf())
 
   it should "execute `1 + 2` == 3" in withInterpreter { interpreter =>
     val response = interpreter.execute("1 + 2")
     response should equal (Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "res0: Int = 3\n"
+      TEXT_PLAIN -> s"${optionalValPrefix}res0: Int = 3\n"
     ))
   }
 
   it should "execute multiple statements" in withInterpreter { interpreter =>
     var response = interpreter.execute("val x = 1")
     response should equal (Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "x: Int = 1\n"
+      TEXT_PLAIN -> s"${optionalValPrefix}x: Int = 1\n"
     ))
 
     response = interpreter.execute("val y = 2")
     response should equal (Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "y: Int = 2\n"
+      TEXT_PLAIN -> s"${optionalValPrefix}y: Int = 2\n"
     ))
 
     response = interpreter.execute("x + y")
     response should equal (Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "res0: Int = 3\n"
+      TEXT_PLAIN -> s"${optionalValPrefix}res0: Int = 3\n"
     ))
   }
 
@@ -63,9 +67,23 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
         |
         |x + y
       """.stripMargin)
-    response should equal(Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "x: Int = 1\ny: Int = 2\nres2: Int = 3\n"
-    ))
+    // The Scala REPL is using an internal counter
+    // whose exact value depends on how the compiler rewrites the block:
+    // Scala 2.12 keeps a per-line counter (so `x`, `y`, then `x + y` yields
+    // `res2`), while Scala 2.13 keeps a per-execution counter (yielding
+    // `res0`). Neither of those is a semantic guarantee -- we just want to
+    // confirm the interpreter bound `x`, `y`, and produced an `Int = 3`
+    // result under *some* `resN` name.
+    val text = response match {
+      case Interpreter.ExecuteSuccess(json) =>
+        (json \\ TEXT_PLAIN).extract[String]
+      case other => fail(s"Expected ExecuteSuccess, got $other")
+    }
+    val expected =
+      s"${optionalValPrefix}x: Int = 1\n" +
+      s"${optionalValPrefix}y: Int = 2\n" +
+      s"${optionalValPrefix}res\\d+: Int = 3\n"
+    text should fullyMatch regex expected
   }
 
   it should "do table magic" in withInterpreter { interpreter =>
@@ -96,7 +114,7 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
       """.stripMargin)
 
     response should equal(Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "res0: Int = 3\n"
+      TEXT_PLAIN -> s"${optionalValPrefix}res0: Int = 3\n"
     ))
   }
 
@@ -132,12 +150,20 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
     val response = interpreter.execute(
       """sc.parallelize(0 to 1).map { i => i+1 }.collect""".stripMargin)
 
-    response should equal(Interpreter.ExecuteSuccess(
-      TEXT_PLAIN -> "res0: Array[Int] = Array(1, 2)\n"
-    ))
+    // 2.13's REPL prints a deprecation banner ahead of the collect result
+    // because `Array` implicit conversion changed; check with contain-like
+    // substrings rather than exact equality so both versions pass.
+    val text = response match {
+      case Interpreter.ExecuteSuccess(json) => (json \\ TEXT_PLAIN).extract[String]
+      case other => fail(s"Expected ExecuteSuccess, got $other")
+    }
+    text should include (s"${optionalValPrefix}res0: Array[Int] = Array(1, 2)")
   }
 
   it should "handle statements ending with comments" in withInterpreter { interpreter =>
+    val expectedResponse =
+      Interpreter.ExecuteSuccess(TEXT_PLAIN -> s"${optionalValPrefix}r: Int = 1\n")
+
     // Test statements with only comments
     var response = interpreter.execute("""// comment""")
     response should equal(Interpreter.ExecuteSuccess(TEXT_PLAIN -> ""))
@@ -154,7 +180,7 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
       """val r = 1
         |// comment
       """.stripMargin)
-    response should equal(Interpreter.ExecuteSuccess(TEXT_PLAIN -> "r: Int = 1\n"))
+    response should equal(expectedResponse)
 
     response = interpreter.execute(
       """val r = 1
@@ -163,7 +189,7 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
         |comment
         |*/
       """.stripMargin)
-    response should equal(Interpreter.ExecuteSuccess(TEXT_PLAIN -> "r: Int = 1\n"))
+    response should equal(expectedResponse)
 
     // Test statements ending with a mix of single line and multi-line comments
     response = interpreter.execute(
@@ -175,7 +201,7 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
         |*/
         |// comment
       """.stripMargin)
-    response should equal(Interpreter.ExecuteSuccess(TEXT_PLAIN -> "r: Int = 1\n"))
+    response should equal(expectedResponse)
 
     response = interpreter.execute(
       """val r = 1
@@ -185,7 +211,7 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
         |comment
         |*/
       """.stripMargin)
-    response should equal(Interpreter.ExecuteSuccess(TEXT_PLAIN -> "r: Int = 1\n"))
+    response should equal(expectedResponse)
 
     // Make sure incomplete statement is still returned as incomplete statement.
     response = interpreter.execute("sc.")
@@ -205,12 +231,14 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
 
     try {
       response should equal(
-        Interpreter.ExecuteSuccess(TEXT_PLAIN -> s"r: String = \n$stringWithComment\n"))
+        Interpreter.ExecuteSuccess(
+          TEXT_PLAIN -> s"${optionalValPrefix}r: String = \n$stringWithComment\n"))
     } catch {
       case _: Exception =>
         response should equal(
-          // Scala 2.11 doesn't have a " " after "="
-          Interpreter.ExecuteSuccess(TEXT_PLAIN -> s"r: String =\n$stringWithComment\n"))
+          // Older Scala versions (2.11) omit the space after `=`.
+          Interpreter.ExecuteSuccess(
+            TEXT_PLAIN -> s"${optionalValPrefix}r: String =\n$stringWithComment\n"))
     }
   }
 
diff --git a/repl/src/test/scala/org/apache/livy/repl/SharedSessionSpec.scala b/repl/src/test/scala/org/apache/livy/repl/SharedSessionSpec.scala
index 18a6db727..3c0b998ff 100644
--- a/repl/src/test/scala/org/apache/livy/repl/SharedSessionSpec.scala
+++ b/repl/src/test/scala/org/apache/livy/repl/SharedSessionSpec.scala
@@ -30,6 +30,9 @@ import org.apache.livy.sessions._
 
 class SharedSessionSpec extends BaseSessionSpec(Shared) {
 
+  // `optionalValPrefix` (Scala-2.13's `val ` before bound-name output, empty on 2.12)
+  // is inherited from `BaseSessionSpec`.
+
   private def execute(session: Session, code: String, codeType: String): Statement = {
     val id = session.execute(code, codeType)
     eventually(timeout(30 seconds), interval(100 millis)) {
@@ -48,7 +51,7 @@ class SharedSessionSpec extends BaseSessionSpec(Shared) {
       "status" -> "ok",
       "execution_count" -> 0,
       "data" -> Map(
-        "text/plain" -> "res0: Int = 3\n"
+        "text/plain" -> s"${optionalValPrefix}res0: Int = 3\n"
       )
     ))
 
@@ -77,16 +80,11 @@ class SharedSessionSpec extends BaseSessionSpec(Shared) {
     statement.id should equal (0)
 
     val result = parse(statement.output)
-
-    val expectedResult = Extraction.decompose(Map(
-      "status" -> "ok",
-      "execution_count" -> 0,
-      "data" -> Map(
-        "text/plain" -> "res0: Array[Int] = Array(1, 2)\n"
-      )
-    ))
-
-    result should equal (expectedResult)
+    // Scala 2.13's REPL prints a deprecation banner before the collect result.
+    val text = ((result \ "data") \ "text/plain").extract[String]
+    text should include (s"${optionalValPrefix}res0: Array[Int] = Array(1, 2)")
+    (result \ "status").extract[String] should equal ("ok")
+    (result \ "execution_count").extract[Int] should equal (0)
   }
 
   it should "throw exception if code type is not specified in shared session" in withSession {
diff --git a/repl/scala-2.12/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala b/repl/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
similarity index 78%
rename from repl/scala-2.12/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
rename to repl/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
index fbad108f4..7b89f9d75 100644
--- a/repl/scala-2.12/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
+++ b/repl/src/test/scala/org/apache/livy/repl/SparkInterpreterSpec.scala
@@ -65,5 +65,20 @@ class SparkInterpreterSpec extends AnyFunSpec with Matchers with LivyBaseUnitTes
       ename shouldBe "java.lang.RuntimeException: message"
       traceback shouldBe expectedTraceback
     }
+
+    it("should skip leading caret lines in Scala 2.13 error format.") {
+      // The 2.13 REPL prints the caret and the offending expression BEFORE
+      // the human-readable "error: ..." line. `parseError` must advance past
+      // the caret line so `ename` still lands on the readable message.
+      val error =
+        """                                ^
+          |error: not found: value abcde
+          |""".stripMargin
+
+      val (ename, traceback) = interpreter.parseError(error)
+      ename shouldBe "error: not found: value abcde"
+      // The pure caret line is retained in the traceback rather than dropped.
+      traceback.exists(_.trim == "^") shouldBe true
+    }
   }
 }
diff --git a/repl/src/test/scala/org/apache/livy/repl/SparkSessionSpec.scala b/repl/src/test/scala/org/apache/livy/repl/SparkSessionSpec.scala
index 90e282839..c28e6b54f 100644
--- a/repl/src/test/scala/org/apache/livy/repl/SparkSessionSpec.scala
+++ b/repl/src/test/scala/org/apache/livy/repl/SparkSessionSpec.scala
@@ -30,6 +30,9 @@ import org.apache.livy.sessions._
 
 class SparkSessionSpec extends BaseSessionSpec(Spark) {
 
+  // `optionalValPrefix` (Scala-2.13's `val ` before bound-name output, empty on 2.12)
+  // is inherited from `BaseSessionSpec`.
+
   it should "execute `1 + 2` == 3" in withSession { session =>
     val statement = execute(session)("1 + 2")
     statement.id should equal (0)
@@ -39,7 +42,7 @@ class SparkSessionSpec extends BaseSessionSpec(Spark) {
       "status" -> "ok",
       "execution_count" -> 0,
       "data" -> Map(
-        "text/plain" -> "res0: Int = 3\n"
+        "text/plain" -> s"${optionalValPrefix}res0: Int = 3\n"
       )
     ))
 
@@ -56,7 +59,7 @@ class SparkSessionSpec extends BaseSessionSpec(Spark) {
       "status" -> "ok",
       "execution_count" -> 0,
       "data" -> Map(
-        "text/plain" -> "x: Int = 1\n"
+        "text/plain" -> s"${optionalValPrefix}x: Int = 1\n"
       )
     ))
 
@@ -70,7 +73,7 @@ class SparkSessionSpec extends BaseSessionSpec(Spark) {
       "status" -> "ok",
       "execution_count" -> 1,
       "data" -> Map(
-        "text/plain" -> "y: Int = 2\n"
+        "text/plain" -> s"${optionalValPrefix}y: Int = 2\n"
       )
     ))
 
@@ -84,7 +87,7 @@ class SparkSessionSpec extends BaseSessionSpec(Spark) {
       "status" -> "ok",
       "execution_count" -> 2,
       "data" -> Map(
-        "text/plain" -> "res0: Int = 3\n"
+        "text/plain" -> s"${optionalValPrefix}res0: Int = 3\n"
       )
     ))
 
@@ -164,16 +167,13 @@ class SparkSessionSpec extends BaseSessionSpec(Spark) {
     statement.id should equal (0)
 
     val result = parse(statement.output)
-
-    val expectedResult = Extraction.decompose(Map(
-      "status" -> "ok",
-      "execution_count" -> 0,
-      "data" -> Map(
-        "text/plain" -> "res0: Array[Int] = Array(1, 2)\n"
-      )
-    ))
-
-    result should equal (expectedResult)
+    // The Scala 2.13 REPL emits a deprecation banner in front of the collect
+    // result (Array-implicit-conversion deprecation). Assert via substring so
+    // both 2.12 (bare line) and 2.13 (banner + line) pass.
+    val text = ((result \ "data") \ "text/plain").extract[String]
+    text should include (s"${optionalValPrefix}res0: Array[Int] = Array(1, 2)")
+    (result \ "status").extract[String] should equal ("ok")
+    (result \ "execution_count").extract[Int] should equal (0)
   }
 
   it should "do table magic" in withSession { session =>
diff --git a/rsc/src/main/java/org/apache/livy/rsc/driver/SparkEntries.java b/rsc/src/main/java/org/apache/livy/rsc/driver/SparkEntries.java
index 6726bb1ab..cc699aed9 100644
--- a/rsc/src/main/java/org/apache/livy/rsc/driver/SparkEntries.java
+++ b/rsc/src/main/java/org/apache/livy/rsc/driver/SparkEntries.java
@@ -24,7 +24,6 @@
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.SparkSession$;
 import org.apache.spark.sql.hive.HiveContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -68,7 +67,7 @@ public SparkSession sparkSession() {
             SparkConf conf = sc().getConf();
             String catalog = conf.get("spark.sql.catalogImplementation", "in-memory").toLowerCase();
 
-            if (catalog.equals("hive") && SparkSession$.MODULE$.hiveClassesArePresent()) {
+            if (catalog.equals("hive") && hiveClassesArePresent()) {
               ClassLoader loader = Thread.currentThread().getContextClassLoader() != null ?
                 Thread.currentThread().getContextClassLoader() : getClass().getClassLoader();
               if (loader.getResource("hive-site.xml") == null) {
@@ -135,4 +134,35 @@ public synchronized void stop() {
       sc.stop();
     }
   }
+
+  /**
+   * Determine whether Spark's Hive support classes are present on the classpath.
+   *
+   * <p>Spark 3 exposed this as {@code org.apache.spark.sql.SparkSession$.hiveClassesArePresent()}.
+   * In Spark 4 {@code SparkSession} became abstract and the concrete companion moved to
+   * {@code org.apache.spark.sql.classic.SparkSession$}. We probe the Spark 4 location first
+   * because on a Spark 4 classpath the old {@code SparkSession$} may still resolve as a stub
+   * that no longer carries the concrete Hive check; the classic companion is the source of
+   * truth. On Spark 3 the classic class is absent and we fall through to the legacy companion.
+   */
+  private static boolean hiveClassesArePresent() {
+    String[] candidates = {
+        "org.apache.spark.sql.classic.SparkSession$", // Spark 4+
+        "org.apache.spark.sql.SparkSession$"          // Spark 3
+    };
+    for (String cls : candidates) {
+      try {
+        Class<?> companion = Class.forName(cls);
+        Object module = companion.getField("MODULE$").get(null);
+        return (Boolean) companion.getMethod("hiveClassesArePresent").invoke(module);
+      } catch (ClassNotFoundException | NoSuchFieldException | NoSuchMethodException e) {
+        // Try the next candidate.
+      } catch (ReflectiveOperationException e) {
+        LOG.warn("Failed to invoke {}.hiveClassesArePresent()", cls, e);
+        return false;
+      }
+    }
+    LOG.warn("Could not locate SparkSession#hiveClassesArePresent on the classpath");
+    return false;
+  }
 }
diff --git a/rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java b/rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java
index 637be8b1d..44bfa654e 100644
--- a/rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java
+++ b/rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java
@@ -193,7 +193,12 @@ public void call(LivyClient client) throws Exception {
         // state changes.
         assertFalse(((JobHandleImpl<Void>)handle).changeState(JobHandle.State.SENT));
 
-        verify(listener).onJobStarted(handle);
+        // Note: onJobStarted is omitted here due to a race condition.
+        // Fast-failing jobs can transition STARTED -> FAILED before addListener()
+        // finishes attaching, causing it to skip onJobStarted and only report FAILED.
+        // Spark 4's faster dispatching makes this more frequent. We only care about
+        // verifying onJobFailed.
+
         verify(listener).onJobFailed(same(handle), any(Throwable.class));
       }
     });
diff --git a/scala-api/scala-2.13/pom.xml b/scala-api/scala-2.13/pom.xml
new file mode 100644
index 000000000..ed44e752a
--- /dev/null
+++ b/scala-api/scala-2.13/pom.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>org.apache.livy</groupId>
+  <artifactId>livy-scala-api_2.13</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <parent>
+    <groupId>org.apache.livy</groupId>
+    <artifactId>livy-scala-api-parent</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+</project>
\ No newline at end of file
diff --git a/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTestUtils.scala b/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTestUtils.scala
index 7d406f3a6..7be5176e8 100644
--- a/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTestUtils.scala
+++ b/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTestUtils.scala
@@ -44,7 +44,7 @@ object ScalaClientTestUtils extends AnyFunSuite with LivyBaseUnitTestSuite {
     for (a <- 1 to count) {
       buffer += r.nextInt()
     }
-    context.sc.parallelize(buffer, partitions).count()
+    context.sc.parallelize(buffer.toSeq, partitions).count()
   }
 
   def assertAwait(lock: CountDownLatch): Unit = {
diff --git a/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaJobHandleTest.scala b/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaJobHandleTest.scala
index 732f6af76..d82282d63 100644
--- a/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaJobHandleTest.scala
+++ b/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaJobHandleTest.scala
@@ -66,9 +66,15 @@ class ScalaJobHandleTest extends AnyFunSuite
   test("ready with Infinite Duration") {
     when(mockJobHandle.isDone).thenReturn(true)
     when(mockJobHandle.get()).thenReturn("hello")
-    val result = Await.ready(scalaJobHandle, Duration.Undefined)
+    // Scala 2.13's Await.ready rejects `Duration.Undefined` as
+    // "Cannot wait for Undefined duration of time"; use `Duration.Inf`.
+    // Also, 2.13's `Await.ready` short-circuits when `isCompleted` is
+    // already true and does not call the underlying `ready(atMost)`, so
+    // the `jobHandle.get()` invocation is not observed on 2.13. Assert
+    // on `isDone` instead, which is exercised by both versions.
+    val result = Await.ready(scalaJobHandle, Duration.Inf)
     assert(result == scalaJobHandle)
-    verify(mockJobHandle, times(1)).get()
+    verify(mockJobHandle, atLeastOnce()).isDone
   }
 
   test("verify addListener call of java jobHandle for onComplete") {
diff --git a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
index de771156e..0d9a0c644 100644
--- a/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/batch/BatchSession.scala
@@ -90,6 +90,11 @@ object BatchSession extends Logging {
 
       builder.redirectOutput(Redirect.PIPE)
       builder.redirectErrorStream(true)
+      // Under unit-test runs we ask spark-submit to be verbose so the tests
+      // can grep the resolved arguments (queue name, master, class, ...) out
+      // of the child's log -- production sessions leave the default off so
+      // the extra output does not pollute the user-facing session log.
+      if (LivyConf.TEST_MODE) builder.verbose(true)
 
       val file = resolveURIs(Seq(request.file), livyConf)(0)
       val sparkSubmit = builder.start(Some(file), request.args)
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
index cfa1c84ee..b193dac80 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSession.scala
@@ -245,18 +245,21 @@ object InteractiveSession extends Logging {
       }
     }
 
-    def datanucleusJars(livyConf: LivyConf, sparkMajorVersion: Int): Seq[String] = {
+    def datanucleusJars(
+        livyConf: LivyConf,
+        sparkMajorVersion: Int,
+        scalaVersion: String): Seq[String] = {
       if (sys.env.getOrElse("LIVY_INTEGRATION_TEST", "false").toBoolean) {
         // datanucleus jars has already been in classpath in integration test
         Seq.empty
       } else {
         val sparkHome = livyConf.sparkHome().get
         val libdir = sparkMajorVersion match {
-          case 3 =>
+          case 3 | 4 =>
             if (new File(sparkHome, "RELEASE").isFile) {
               new File(sparkHome, "jars")
             } else {
-              new File(sparkHome, "assembly/target/scala-2.12/jars")
+              new File(sparkHome, s"assembly/target/scala-$scalaVersion/jars")
             }
           case v =>
             throw new RuntimeException(
@@ -340,17 +343,19 @@ object InteractiveSession extends Logging {
       }
     }
 
-    def mergeHiveSiteAndHiveDeps(sparkMajorVersion: Int): Unit = {
+    def mergeHiveSiteAndHiveDeps(sparkMajorVersion: Int, scalaVersion: String): Unit = {
       val sparkFiles = conf.get("spark.files").map(_.split(",")).getOrElse(Array.empty[String])
       hiveSiteFile(sparkFiles, livyConf) match {
         case (_, true) =>
           debug("Enable HiveContext because hive-site.xml is found in user request.")
-          mergeConfList(datanucleusJars(livyConf, sparkMajorVersion), LivyConf.SPARK_JARS)
+          mergeConfList(
+            datanucleusJars(livyConf, sparkMajorVersion, scalaVersion), LivyConf.SPARK_JARS)
         case (Some(file), false) =>
           debug("Enable HiveContext because hive-site.xml is found under classpath, "
             + file.getAbsolutePath)
           mergeConfList(List(file.getAbsolutePath), LivyConf.SPARK_FILES)
-          mergeConfList(datanucleusJars(livyConf, sparkMajorVersion), LivyConf.SPARK_JARS)
+          mergeConfList(
+            datanucleusJars(livyConf, sparkMajorVersion, scalaVersion), LivyConf.SPARK_JARS)
         case (None, false) =>
           warn("Enable HiveContext but no hive-site.xml found under" +
             " classpath or user request.")
@@ -396,7 +401,7 @@ object InteractiveSession extends Logging {
     builderProperties.put("spark.sql.catalogImplementation", confVal)
 
     if (enableHiveContext) {
-      mergeHiveSiteAndHiveDeps(sparkMajorVersion)
+      mergeHiveSiteAndHiveDeps(sparkMajorVersion, scalaVersion)
     }
 
     // Pick all the RSC-specific configs that have not been explicitly set otherwise, and
diff --git a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
index d6ba9e9df..a39df3bcb 100644
--- a/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
+++ b/server/src/main/scala/org/apache/livy/server/interactive/InteractiveSessionServlet.scala
@@ -118,7 +118,11 @@ class InteractiveSessionServlet(
 
     new SessionInfo(session.id, session.name.orNull, session.appId.orNull,
       session.owner, session.state.toString, session.kind.toString,
-      session.appInfo.asJavaMap, logs.asJava, session.ttl.orNull,
+      // `.toSeq.asJava`: `SessionInfo` needs a `java.util.List`. In Scala
+      // 2.13 the JavaConverters `.asJava` overload that produces a
+      // `java.util.List` is the `Seq[A]` one; forcing `.toSeq` here picks
+      // that overload regardless of which concrete collection `logs` is.
+      session.appInfo.asJavaMap, logs.toSeq.asJava, session.ttl.orNull,
       session.idleTimeout.orNull, session.driverMemory.orNull,
       session.driverCores.getOrElse(0), session.executorMemory.orNull,
       session.executorCores.getOrElse(0), conf, archives,
diff --git a/server/src/main/scala/org/apache/livy/utils/LivySparkUtils.scala b/server/src/main/scala/org/apache/livy/utils/LivySparkUtils.scala
index 6bd8c3807..0a431819c 100644
--- a/server/src/main/scala/org/apache/livy/utils/LivySparkUtils.scala
+++ b/server/src/main/scala/org/apache/livy/utils/LivySparkUtils.scala
@@ -30,6 +30,10 @@ object LivySparkUtils extends Logging {
   // For each Spark version we supported, we need to add this mapping relation in case Scala
   // version cannot be detected from "spark-submit --version".
   private val _defaultSparkScalaVersion = SortedMap(
+    // Spark 4.1 + Scala 2.13
+    (4, 1) -> "2.13",
+    // Spark 4.0 + Scala 2.13
+    (4, 0) -> "2.13",
     // Spark 3.5 + Scala 2.12
     (3, 5) -> "2.12",
     // Spark 3.4 + Scala 2.12
@@ -46,7 +50,7 @@ object LivySparkUtils extends Logging {
 
   // Supported Spark version (Spark 2.x support has been removed)
   private val MIN_VERSION = (3, 0)
-  private val MAX_VERSION = (3, 6)
+  private val MAX_VERSION = (4, 2)
 
   private val sparkVersionRegex = """version (.*)""".r.unanchored
   private val scalaVersionRegex = """Scala version (.*), Java""".r.unanchored
diff --git a/server/src/main/scala/org/apache/livy/utils/SparkProcessBuilder.scala b/server/src/main/scala/org/apache/livy/utils/SparkProcessBuilder.scala
index 01cbb4c3c..3c9b45bf6 100644
--- a/server/src/main/scala/org/apache/livy/utils/SparkProcessBuilder.scala
+++ b/server/src/main/scala/org/apache/livy/utils/SparkProcessBuilder.scala
@@ -38,6 +38,7 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
   private[this] var _redirectOutput: Option[ProcessBuilder.Redirect] = None
   private[this] var _redirectError: Option[ProcessBuilder.Redirect] = None
   private[this] var _redirectErrorStream: Option[Boolean] = None
+  private[this] var _verbose: Boolean = false
 
   def executable(executable: String): SparkProcessBuilder = {
     _executable = executable
@@ -154,6 +155,16 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
     this
   }
 
+  /**
+   * Enable `spark-submit --verbose`. When set, spark-submit prints its
+   * parsed argument list (queue, master, class, etc.) to stderr; this is
+   * primarily useful for tests that need to assert on the resolved arguments.
+   */
+  def verbose(v: Boolean): SparkProcessBuilder = {
+    _verbose = v
+    this
+  }
+
   def start(file: Option[String], args: Traversable[String]): LineBufferedProcess = {
     var arguments = ArrayBuffer(_executable)
 
@@ -192,6 +203,8 @@ class SparkProcessBuilder(livyConf: LivyConf) extends Logging {
 
     addOpt("--queue", _queue)
 
+    if (_verbose) arguments += "--verbose"
+
     arguments += file.getOrElse("spark-internal")
     arguments ++= args
 
diff --git a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
index 556c54fbd..1efa266ac 100644
--- a/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
+++ b/server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala
@@ -201,12 +201,15 @@ class InteractiveSessionSpec extends AnyFunSpec
         "data" -> Map("text/plain" -> "3")))
       )
 
+      // The Scala REPL under Spark 4 (2.13) may or may not prefix the
+      // bound-name output with `val ` depending on the driver JVM's precise
+      // `-Yrepl-class-based` handling; accept either form so both spark3 and
+      // spark4 builds pass without special-casing environment differences.
       val scalaResult = executeStatement("1 + 2", Some("spark"))
-      scalaResult should equal (Extraction.decompose(Map(
-        "status" -> "ok",
-        "execution_count" -> 1,
-        "data" -> Map("text/plain" -> "res0: Int = 3\n")))
-      )
+      val scalaData = ((scalaResult \ "data") \ "text/plain").extract[String]
+      scalaData should (equal ("res0: Int = 3\n") or equal ("val res0: Int = 3\n"))
+      (scalaResult \ "status").extract[String] should equal ("ok")
+      (scalaResult \ "execution_count").extract[Int] should equal (1)
 
       val rResult = executeStatement("1 + 2", Some("sparkr"))
       rResult should equal (Extraction.decompose(Map(
diff --git a/server/src/test/scala/org/apache/livy/utils/LivySparkUtilsSuite.scala b/server/src/test/scala/org/apache/livy/utils/LivySparkUtilsSuite.scala
index 4338f6d38..d918adab6 100644
--- a/server/src/test/scala/org/apache/livy/utils/LivySparkUtilsSuite.scala
+++ b/server/src/test/scala/org/apache/livy/utils/LivySparkUtilsSuite.scala
@@ -48,6 +48,8 @@ class LivySparkUtilsSuite extends AnyFunSuite with Matchers with LivyBaseUnitTes
     testSparkVersion("3.1.0")
     testSparkVersion("3.2.0")
     testSparkVersion("3.5.0")
+    testSparkVersion("4.0.0")
+    testSparkVersion("4.1.2")
   }
 
   test("should complain about unsupported Spark versions") {
@@ -91,6 +93,8 @@ class LivySparkUtilsSuite extends AnyFunSuite with Matchers with LivyBaseUnitTes
     defaultSparkScalaVersion(formatSparkVersion("3.0.0")) shouldBe "2.12"
     defaultSparkScalaVersion(formatSparkVersion("3.1.0")) shouldBe "2.12"
     defaultSparkScalaVersion(formatSparkVersion("3.5.0")) shouldBe "2.12"
+    defaultSparkScalaVersion(formatSparkVersion("4.0.0")) shouldBe "2.13"
+    defaultSparkScalaVersion(formatSparkVersion("4.1.2")) shouldBe "2.13"
   }
 
   test("sparkScalaVersion() should use spark-submit detected Scala version.") {
diff --git a/test-lib/src/main/scala/org/apache/livy/test/ScalaVersionAware.scala b/test-lib/src/main/scala/org/apache/livy/test/ScalaVersionAware.scala
new file mode 100644
index 000000000..97e39d214
--- /dev/null
+++ b/test-lib/src/main/scala/org/apache/livy/test/ScalaVersionAware.scala
@@ -0,0 +1,50 @@
+/*
+ * 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.livy.test
+
+/**
+ * Shared Scala-version-dependent helpers for the Livy test suites. Mixed into
+ * base spec classes so their subclasses see a single canonical
+ * `optionalValPrefix` (and `optional*PrefixRegex` fragments for regex-based
+ * `verifyResult` calls in integration tests). Detection uses the runtime
+ * Scala version, so this trait works regardless of which Scala the artifact
+ * was compiled against.
+ */
+trait ScalaVersionAware {
+
+  /** Prefix that the Scala 2.13 REPL prints before a value binding, e.g.
+   *  `val res0: Int = 2` vs. Scala 2.12's `res0: Int = 2`. Empty on 2.12.
+   *  Use in exact-match expected strings. */
+  protected val optionalValPrefix: String =
+    if (scala.util.Properties.versionNumberString.startsWith("2.13")) "val " else ""
+
+  /** Regex fragment `(val )?` -- an optional-`val ` alternation that matches
+   *  the Scala 2.12 and 2.13 REPL output shape in one pattern. Use in regex-
+   *  based `verifyResult` calls. */
+  protected val optionalValPrefixRegex: String = "(val )?"
+
+  /** Regex fragment `(?:defined )?` -- Scala 2.12's `defined class X` vs.
+   *  Scala 2.13's plain `class X` REPL output. */
+  protected val optionalDefinedPrefixRegex: String = "(?:defined )?"
+
+  /** Regex fragment `(?:warning:.*\n)?` -- Scala 2.13 emits a
+   *  `warning: n deprecation(s)...` line above value bindings that use a
+   *  deprecated API; Scala 2.12 does not. Pair with `(?s)` DOTALL to span
+   *  the newline. */
+  protected val optionalWarningPrefixRegex: String = "(?:warning:.*\\n)?"
+}
diff --git a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyExecuteStatementOperation.scala b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyExecuteStatementOperation.scala
index f7d6c16b0..234d21acf 100644
--- a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyExecuteStatementOperation.scala
+++ b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/LivyExecuteStatementOperation.scala
@@ -205,6 +205,7 @@ class LivyExecuteStatementOperation(
     }
     val res = new mutable.ListBuffer[String]
     while (fetchNext(res)) {}
-    res
+    // Scala 2.13 no longer widens a mutable Buffer to Seq implicitly.
+    res.toSeq
   }
 }
diff --git a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/cli/ThriftBinaryCLIService.scala b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/cli/ThriftBinaryCLIService.scala
index 72b693018..d5a46cac1 100644
--- a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/cli/ThriftBinaryCLIService.scala
+++ b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/cli/ThriftBinaryCLIService.scala
@@ -108,11 +108,17 @@ class ThriftBinaryCLIService(override val cliService: LivyCLIService, val oomHoo
         .protocolFactory(new TBinaryProtocol.Factory)
         .inputProtocolFactory(
           new TBinaryProtocol.Factory(true, true, maxMessageSize, maxMessageSize))
-        .requestTimeout(requestTimeout)
-        .requestTimeoutUnit(TimeUnit.MILLISECONDS)
-        .beBackoffSlotLength(beBackoffSlotLength)
-        .beBackoffSlotLengthUnit(TimeUnit.MILLISECONDS)
         .executorService(executorService)
+      // `requestTimeout`, `requestTimeoutUnit`, `beBackoffSlotLength` and
+      // `beBackoffSlotLengthUnit` are present in the libthrift that Livy
+      // compiles against (0.9.3, pinned in the root pom) and in the versions
+      // Spark 3 brings in transitively, but were removed in libthrift 0.16.0
+      // (transitively pulled in by Spark 4). Invoke them reflectively so the
+      // same code compiles and runs on both classpaths.
+      applyOptionalArg(sargs, "requestTimeout", Integer.TYPE, Int.box(requestTimeout))
+      applyOptionalArg(sargs, "requestTimeoutUnit", classOf[TimeUnit], TimeUnit.MILLISECONDS)
+      applyOptionalArg(sargs, "beBackoffSlotLength", Integer.TYPE, Int.box(beBackoffSlotLength))
+      applyOptionalArg(sargs, "beBackoffSlotLengthUnit", classOf[TimeUnit], TimeUnit.MILLISECONDS)
       // TCP Server
       server = new TThreadPoolServer(sargs)
       server.setServerEventHandler(new TServerEventHandler() {
@@ -174,4 +180,26 @@ class ThriftBinaryCLIService(override val cliService: LivyCLIService, val oomHoo
     server = null
     info("Thrift server has stopped")
   }
+
+  /**
+   * Invoke a `TThreadPoolServer.Args` builder method by name if it exists on
+   * the runtime libthrift version. Silently skip it if the method is missing:
+   * libthrift 0.16.0 (pulled in by the Spark 4 profile via spark-hive)
+   * dropped the login-timeout / backoff-slot builders that pre-0.16 releases
+   * -- including Livy's pinned 0.9.3 compile dependency -- still expose.
+   */
+  private def applyOptionalArg(
+      args: TThreadPoolServer.Args,
+      methodName: String,
+      paramType: Class[_],
+      value: AnyRef): Unit = {
+    try {
+      val m = classOf[TThreadPoolServer.Args].getMethod(methodName, paramType)
+      m.invoke(args, value)
+    } catch {
+      case _: NoSuchMethodException =>
+        debug(s"TThreadPoolServer.Args.$methodName not present in the runtime " +
+          s"libthrift; skipping (removed in libthrift 0.16.0).")
+    }
+  }
 }
diff --git a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/types/DataTypeUtils.scala b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/types/DataTypeUtils.scala
index f8f0f190d..ce898d82c 100644
--- a/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/types/DataTypeUtils.scala
+++ b/thriftserver/server/src/main/scala/org/apache/livy/thriftserver/types/DataTypeUtils.scala
@@ -17,7 +17,7 @@
 
 package org.apache.livy.thriftserver.types
 
-import org.json4s.{DefaultFormats, JValue, StringInput}
+import org.json4s.{DefaultFormats, JValue}
 import org.json4s.JsonAST.{JObject, JString}
 import org.json4s.jackson.JsonMethods.parse
 
@@ -76,7 +76,10 @@ object DataTypeUtils {
    * @return a [[Schema]] representing the schema provided as input
    */
   def schemaFromSparkJson(sparkJson: String): Schema = {
-    val schema = parse(StringInput(sparkJson), false) \ "fields"
+    // json4s 4.x switched `parse(input, useBigDecimalForDouble)` to require
+    // an implicit AsJsonInput[T]; `parse(sparkJson)` picks the String
+    // overload directly and behaves identically for our purpose.
+    val schema = parse(sparkJson) \ "fields"
     val fields = schema.children.map { field =>
       val name = (field \ "name").extract[String]
       val hiveType = toFieldType(field \ "type")
diff --git a/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerSuites.scala b/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerSuites.scala
index dd688a64a..a8672090e 100644
--- a/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerSuites.scala
+++ b/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerSuites.scala
@@ -570,8 +570,13 @@ class BinaryThriftServerSuite extends ThriftServerBaseTest with CommonThriftTest
         }
       }
       val message = caught.getMessage
-      assert(message.contains("Database 'invalid_database' not found") ||
-        message.contains("The schema `invalid_database` cannot be found"))
+      // Spark 3: "Database 'invalid_database' not found"
+      // Spark 3.3+: "The schema `invalid_database` cannot be found"
+      // Spark 4: "[SCHEMA_NOT_FOUND] The schema `spark_catalog`.`invalid_database` cannot be found"
+      assert(message.contains("invalid_database") && (
+        message.contains("not found") ||
+        message.contains("cannot be found") ||
+        message.contains("SCHEMA_NOT_FOUND")))
     }
   }
 
diff --git a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetColumnsJob.java b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetColumnsJob.java
index b5a0cba7a..7e4214f20 100644
--- a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetColumnsJob.java
+++ b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetColumnsJob.java
@@ -20,7 +20,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static scala.collection.JavaConversions.seqAsJavaList;
+import static scala.collection.JavaConverters.seqAsJavaListConverter;
 
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.TableIdentifier;
@@ -50,14 +50,15 @@ public GetColumnsJob(
   @Override
   protected List<Row> fetchCatalogObjects(SessionCatalog catalog) {
     List<Row> columnList = new ArrayList<>();
-    List<String> databases = seqAsJavaList(catalog.listDatabases(databasePattern));
+    List<String> databases =
+            seqAsJavaListConverter(catalog.listDatabases(databasePattern)).asJava();
 
     for (String db : databases) {
       List<TableIdentifier> tableIdentifiers =
-        seqAsJavaList(catalog.listTables(db, tablePattern));
+        seqAsJavaListConverter(catalog.listTables(db, tablePattern)).asJava();
       for (TableIdentifier tableIdentifier : tableIdentifiers) {
         CatalogTable table = catalog.getTempViewOrPermanentTableMetadata(tableIdentifier);
-        List<StructField> fields = seqAsJavaList(table.schema());
+        List<StructField> fields = seqAsJavaListConverter(table.schema()).asJava();
         int position = 0;
         for (StructField field : fields) {
           if (field.name().matches(columnPattern)) {
diff --git a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetFunctionsJob.java b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetFunctionsJob.java
index e5e383aee..db5bf5eb8 100644
--- a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetFunctionsJob.java
+++ b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetFunctionsJob.java
@@ -21,7 +21,7 @@
 import java.util.List;
 
 import scala.Tuple2;
-import static scala.collection.JavaConversions.seqAsJavaList;
+import static scala.collection.JavaConverters.seqAsJavaListConverter;
 
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.FunctionIdentifier;
@@ -51,10 +51,11 @@ public GetFunctionsJob(
   protected List<Row> fetchCatalogObjects(SessionCatalog catalog) {
     List<Row> funcList = new ArrayList<>();
 
-    List<String> databases = seqAsJavaList(catalog.listDatabases(databasePattern));
+    List<String> databases =
+            seqAsJavaListConverter(catalog.listDatabases(databasePattern)).asJava();
     for (String db : databases) {
       List<Tuple2<FunctionIdentifier, String>> identifiersTypes =
-        seqAsJavaList(catalog.listFunctions(db, functionRegex));
+        seqAsJavaListConverter(catalog.listFunctions(db, functionRegex)).asJava();
       for (Tuple2<FunctionIdentifier, String> identifierType : identifiersTypes) {
         FunctionIdentifier function = identifierType._1;
         ExpressionInfo info = catalog.lookupFunctionInfo(function);
diff --git a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetSchemasJob.java b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetSchemasJob.java
index 59c4ccace..25a723ec8 100644
--- a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetSchemasJob.java
+++ b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetSchemasJob.java
@@ -20,7 +20,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static scala.collection.JavaConversions.seqAsJavaList;
+import static scala.collection.JavaConverters.seqAsJavaListConverter;
 
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.catalog.SessionCatalog;
@@ -40,7 +40,8 @@ public GetSchemasJob(
 
   @Override
   protected List<Row> fetchCatalogObjects(SessionCatalog catalog) {
-    List<String> databases = seqAsJavaList(catalog.listDatabases(schemaPattern));
+    List<String> databases =
+        seqAsJavaListConverter(catalog.listDatabases(schemaPattern)).asJava();
     List<Row> schemas = new ArrayList<>();
     for (String db : databases) {
       schemas.add(new GenericRow(new Object[] {
diff --git a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetTablesJob.java b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetTablesJob.java
index d3c6b5363..e45ffcb4f 100644
--- a/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetTablesJob.java
+++ b/thriftserver/session/src/main/java/org/apache/livy/thriftserver/session/GetTablesJob.java
@@ -20,7 +20,7 @@
 import java.util.ArrayList;
 import java.util.List;
 
-import static scala.collection.JavaConversions.seqAsJavaList;
+import static scala.collection.JavaConverters.seqAsJavaListConverter;
 
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.TableIdentifier;
@@ -54,10 +54,11 @@ public GetTablesJob(
   @Override
   protected List<Row> fetchCatalogObjects(SessionCatalog catalog) {
     List<Row> tableList = new ArrayList<Row>();
-    List<String> databases = seqAsJavaList(catalog.listDatabases(databasePattern));
+    List<String> databases =
+            seqAsJavaListConverter(catalog.listDatabases(databasePattern)).asJava();
     for (String db : databases) {
       List<TableIdentifier> tableIdentifiers =
-        seqAsJavaList(catalog.listTables(db, tablePattern));
+        seqAsJavaListConverter(catalog.listTables(db, tablePattern)).asJava();
       for (TableIdentifier tableIdentifier : tableIdentifiers) {
         CatalogTable table = catalog.getTempViewOrPermanentTableMetadata(tableIdentifier);
         String type = convertTableType(table.tableType().name());