This is an automated email from the ASF dual-hosted git repository.
gyogal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/livy.git
The following commit(s) were added to refs/heads/master by this push:
new d0821740 [LIVY-1064] Initial changes to provide Scala 2.13
compatibility
d0821740 is described below
commit d0821740848c2b2d3f9bdf084b3b457665a25984
Author: idzikovsky <[email protected]>
AuthorDate: Thu Aug 6 04:38:26 2026 +0300
[LIVY-1064] Initial changes to provide Scala 2.13 compatibility
## What changes were proposed in this pull request?
I've stared working on Spark 4 compatibility in my local fork. I don't
think that all changes that I've made is applicable to be pushed to the
upstream project, but this change I'm confident about.
In this fix I mainly updated usage of collections to be compatible with new
redesigned Scala 2.13 collections (so as with older Scala 2.11/2.12
collections).
Also, I replaced invocation of `ServletContext.initParameters()` with
`ServletContext.setInitParameter()`, because the old wrapper is not compatible
with a new `jakarta.servlet.ServletContext` interface (Spark 4 migrated from
javax to jakarta), while `ServletContext.setInitParameter()` is
version-independent.
## How was this patch tested?
I run unit tests and manually tested that I'm able to create a Session.
## Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8.
---
repl/src/main/scala/org/apache/livy/repl/Session.scala | 2 +-
.../src/main/scala/org/apache/livy/scalaapi/package.scala | 2 +-
.../src/main/scala/org/apache/livy/server/LivyServer.scala | 2 +-
.../apache/livy/server/interactive/InteractiveSession.scala | 6 +++---
.../livy/server/interactive/InteractiveSessionServlet.scala | 2 +-
.../org/apache/livy/server/recovery/ZooKeeperManager.scala | 2 +-
.../scala/org/apache/livy/utils/SparkKubernetesApp.scala | 12 ++++++------
.../src/main/scala/org/apache/livy/utils/SparkYarnApp.scala | 9 ++++-----
8 files changed, 18 insertions(+), 19 deletions(-)
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 262c811c..c1267bc4 100644
--- a/repl/src/main/scala/org/apache/livy/repl/Session.scala
+++ b/repl/src/main/scala/org/apache/livy/repl/Session.scala
@@ -137,7 +137,7 @@ class Session(
entries
}(interpreterExecutor)
- future.onFailure { case _ => changeState(SessionState.Error())
}(interpreterExecutor)
+ future.failed.foreach { _ => changeState(SessionState.Error())
}(interpreterExecutor)
future
}
diff --git a/scala-api/src/main/scala/org/apache/livy/scalaapi/package.scala
b/scala-api/src/main/scala/org/apache/livy/scalaapi/package.scala
index a08c7002..a4914b05 100644
--- a/scala-api/src/main/scala/org/apache/livy/scalaapi/package.scala
+++ b/scala-api/src/main/scala/org/apache/livy/scalaapi/package.scala
@@ -42,7 +42,7 @@ package object scalaapi {
private[livy] def getJavaFutureResult[T](jFuture: JFuture[T],
atMost: Duration =
Duration.Undefined): T = {
try {
- if (!atMost.isFinite()) jFuture.get else jFuture.get(atMost.toMillis,
TimeUnit.MILLISECONDS)
+ if (!atMost.isFinite) jFuture.get else jFuture.get(atMost.toMillis,
TimeUnit.MILLISECONDS)
} catch {
case executionException: ExecutionException => throw
executionException.getCause
}
diff --git a/server/src/main/scala/org/apache/livy/server/LivyServer.scala
b/server/src/main/scala/org/apache/livy/server/LivyServer.scala
index 7b6f323e..73f259d5 100644
--- a/server/src/main/scala/org/apache/livy/server/LivyServer.scala
+++ b/server/src/main/scala/org/apache/livy/server/LivyServer.scala
@@ -229,7 +229,7 @@ class LivyServer extends Logging {
override def contextInitialized(sce: ServletContextEvent): Unit = {
try {
val context = sce.getServletContext()
- context.initParameters(org.scalatra.EnvironmentKey) =
livyConf.get(ENVIRONMENT)
+ context.setInitParameter(org.scalatra.EnvironmentKey,
livyConf.get(ENVIRONMENT))
val interactiveServlet = new InteractiveSessionServlet(
interactiveSessionManager, sessionStore, livyConf, accessManager)
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 6af80133..cfa1c84e 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
@@ -485,15 +485,15 @@ class InteractiveSession(
client.get.getServerUri.get()
}(sessionManageExecutors)
- uriFuture.onSuccess { case url =>
+ uriFuture.foreach { url =>
rscDriverUri = Option(url)
sessionSaveLock.synchronized {
sessionStore.save(RECOVERY_SESSION_TYPE, recoveryMetadata)
}
}(sessionManageExecutors)
- uriFuture.onFailure {
- case e => warn("Fail to get rsc uri", e)
+ uriFuture.failed.foreach { e =>
+ warn("Fail to get rsc uri", e)
}(sessionManageExecutors)
// Send a dummy job that will return once the client is ready to be
used, and set the
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 310504e7..d6ba9e9d 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
@@ -89,7 +89,7 @@ class InteractiveSessionServlet(
val from = math.max(0, lines.length - size)
val until = from + size
- lines.view(from, until)
+ lines.slice(from, until)
}
.getOrElse(Nil)
} else {
diff --git
a/server/src/main/scala/org/apache/livy/server/recovery/ZooKeeperManager.scala
b/server/src/main/scala/org/apache/livy/server/recovery/ZooKeeperManager.scala
index ff483e5d..9ad86bdc 100644
---
a/server/src/main/scala/org/apache/livy/server/recovery/ZooKeeperManager.scala
+++
b/server/src/main/scala/org/apache/livy/server/recovery/ZooKeeperManager.scala
@@ -169,7 +169,7 @@ class ZooKeeperManager(
if (curatorClient.checkExists().forPath(key) == null) {
Seq.empty[String]
} else {
- curatorClient.getChildren.forPath(key).asScala
+ curatorClient.getChildren.forPath(key).asScala.toSeq
}
}
diff --git
a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala
b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala
index 22ab6b8f..9c813edc 100644
--- a/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala
+++ b/server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala
@@ -278,7 +278,7 @@ class SparkKubernetesApp private[utils] (
"Please check Livy log and KUBERNETES log to know the details."
error(s"Failed monitoring the app $appTag: $msg")
- kubernetesDiagnostics = ArrayBuffer(msg)
+ kubernetesDiagnostics = IndexedSeq(msg)
failToMonitor()
}
}
@@ -360,12 +360,12 @@ class SparkKubernetesApp private[utils] (
kubernetesAppMonitorFailedTimes += 1
if (kubernetesAppMonitorFailedTimes > appLookupMaxFailedTimes) {
error(s"Monitoring of the app $appTag was interrupted.", e)
- kubernetesDiagnostics = ArrayBuffer(e.getMessage)
+ kubernetesDiagnostics = IndexedSeq(e.getMessage)
failToMonitor()
}
case NonFatal(e) =>
error(s"Error while refreshing Kubernetes state", e)
- kubernetesDiagnostics = ArrayBuffer(e.getMessage)
+ kubernetesDiagnostics = IndexedSeq(e.getMessage)
changeState(SparkApp.State.FAILED)
} finally {
if (!isRunning) {
@@ -636,7 +636,7 @@ private[utils] case class KubernetesAppReport(driver:
Option[Pod], executors: Se
private def buildSparkPodDiagnosticsPrettyString(pod: Pod): String = {
import scala.collection.JavaConverters._
- def printMap(map: Map[_, _]): String = map.map {
+ def printMap[K, V](map: Map[K, V]): String = map.map {
case (key, value) => s"$key=$value"
}.mkString(", ")
@@ -690,7 +690,7 @@ private[utils] object KubernetesExtensions {
.withLabels(labels.asJava)
.withLabel(appTagLabel)
.withLabel(appIdLabel)
- .list.getItems.asScala.map(new KubernetesApplication(_))
+ .list.getItems.asScala.map(new KubernetesApplication(_)).toSeq
}
def killApplication(app: KubernetesApplication): Boolean = {
@@ -705,7 +705,7 @@ private[utils] object KubernetesExtensions {
): KubernetesAppReport = {
val pods = client.pods.inNamespace(app.getApplicationNamespace)
.withLabels(Map(appTagLabel -> app.getApplicationTag).asJava)
- .list.getItems.asScala
+ .list.getItems.asScala.toSeq
val driver = pods.find(_.getMetadata.getLabels.get(SPARK_ROLE_LABEL) ==
SPARK_ROLE_DRIVER)
val executors =
pods.filter(_.getMetadata.getLabels.get(SPARK_ROLE_LABEL) ==
SPARK_ROLE_EXECUTOR)
diff --git a/server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala
b/server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala
index 35895084..1ff40726 100644
--- a/server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala
+++ b/server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala
@@ -20,7 +20,6 @@ import java.util
import scala.annotation.tailrec
import scala.collection.JavaConverters._
-import scala.collection.mutable.ArrayBuffer
import scala.concurrent._
import scala.concurrent.duration._
import scala.language.postfixOps
@@ -155,8 +154,8 @@ class SparkYarnApp private[utils] (
private var yarnDiagnostics: IndexedSeq[String] = IndexedSeq.empty[String]
override def log(): IndexedSeq[String] =
- ("stdout: " +:
process.map(_.inputLines).getOrElse(ArrayBuffer.empty[String])) ++
- ("\nstderr: " +:
process.map(_.errorLines).getOrElse(ArrayBuffer.empty[String])) ++
+ ("stdout: " +:
process.map(_.inputLines).getOrElse(IndexedSeq.empty[String])) ++
+ ("\nstderr: " +:
process.map(_.errorLines).getOrElse(IndexedSeq.empty[String])) ++
("\nYARN Diagnostics: " +: yarnDiagnostics)
override def kill(): Unit = synchronized {
@@ -360,11 +359,11 @@ class SparkYarnApp private[utils] (
debug(s"$appId $state ${yarnDiagnostics.mkString(" ")}")
} catch {
case _: InterruptedException =>
- yarnDiagnostics = ArrayBuffer("Session stopped by user.")
+ yarnDiagnostics = IndexedSeq("Session stopped by user.")
changeState(SparkApp.State.KILLED)
case NonFatal(e) =>
error(s"Error whiling refreshing YARN state", e)
- yarnDiagnostics = ArrayBuffer(e.getMessage)
+ yarnDiagnostics = IndexedSeq(e.getMessage)
changeState(SparkApp.State.FAILED)
}
}
