[LIVY-1086][TEST] Pin Spark driver to loopback in remaining macOS test fixtures

client-common/src/main/java/org/apache/livy/client/common/TestUtils.java

@@ -40,6 +40,15 @@ public class TestUtils {
    */
   public static final String TEST_BIND_HOST = "127.0.0.1";
 
+  /**
+   * Spark configuration keys used by tests to pin the Spark driver to a locally reachable
+   * (loopback) address. Centralized here so the raw config strings are not duplicated across
+   * the test suites of every module (see LIVY-1065). Both are typically set to
+   * {@link #TEST_BIND_HOST}.
+   */
+  public static final String SPARK_DRIVER_HOST = "spark.driver.host";
+  public static final String SPARK_DRIVER_BIND_ADDRESS = "spark.driver.bindAddress";
+
   /**
    * Returns JVM arguments that enable jacoco on a process to be run. The returned arguments
    * create a new, unique output file in the same directory referenced by the "jacoco.args"

integration-test/src/main/scala/org/apache/livy/test/framework/MiniCluster.scala

@@ -241,7 +241,7 @@ class MiniCluster(config: Map[String, String]) extends Cluster with MiniClusterU
       "spark.executor.instances" -> "1",
       "spark.scheduler.minRegisteredResourcesRatio" -> "0.0",
       "spark.ui.enabled" -> "false",
-      "spark.driver.host" -> TestUtils.TEST_BIND_HOST,
+      TestUtils.SPARK_DRIVER_HOST -> TestUtils.TEST_BIND_HOST,
       "spark.yarn.appMasterEnv.SPARK_LOCAL_IP" -> TestUtils.TEST_BIND_HOST,
       "spark.executorEnv.SPARK_LOCAL_IP" -> TestUtils.TEST_BIND_HOST,
       // Propagate the host shell PATH into the YARN AM and executor

repl/src/test/scala/org/apache/livy/repl/BaseSessionSpec.scala

@@ -49,6 +49,8 @@ abstract class BaseSessionSpec(kind: Kind)
   }
 
   private val sparkConf = new SparkConf()
+    .set(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+    .set(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
 
   protected def execute(session: Session)(code: String): Statement = {
     val id = session.execute(code)

repl/src/test/scala/org/apache/livy/repl/PythonInterpreterSpec.scala

@@ -25,6 +25,7 @@ import org.scalatest.Inside.inside
 import org.scalatest.funspec.AnyFunSpec
 import org.scalatest.matchers.should.Matchers
 
+import org.apache.livy.client.common.TestUtils
 import org.apache.livy.rsc.driver.SparkEntries
 import org.apache.livy.sessions._
 
@@ -301,6 +302,8 @@ class PythonInterpreterSpec extends PythonBaseInterpreterSpec with BeforeAndAfte
 
   override def createInterpreter(): Interpreter = {
     val sparkConf = new SparkConf()
+      .set(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+      .set(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
     PythonInterpreter(sparkConf, new SparkEntries(sparkConf))
   }
 

repl/src/test/scala/org/apache/livy/repl/ReplDriverSuite.scala

@@ -45,6 +45,8 @@ class ReplDriverSuite extends AnyFunSuite with LivyBaseUnitTestSuite {
       .setConf(SparkLauncher.EXECUTOR_EXTRA_CLASSPATH, sys.props("java.class.path"))
       .setConf(RSCConf.Entry.LIVY_JARS.key(), "")
       .setConf(RSCConf.Entry.RPC_SERVER_ADDRESS.key(), TestUtils.TEST_BIND_HOST)
+      .setConf(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+      .setConf(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
       .setURI(new URI("rsc:/"))
       .setConf(RSCConf.Entry.DRIVER_CLASS.key(), classOf[ReplDriver].getName())
       .setConf(RSCConf.Entry.SESSION_KIND.key(), Spark.toString)

repl/src/test/scala/org/apache/livy/repl/SQLInterpreterSpec.scala

@@ -26,6 +26,7 @@ import org.json4s.{DefaultFormats, JValue}
 import org.json4s.JsonAST.{JArray, JNull}
 import org.json4s.JsonDSL._
 
+import org.apache.livy.client.common.TestUtils
 import org.apache.livy.rsc.RSCConf
 import org.apache.livy.rsc.driver.SparkEntries
 
@@ -40,6 +41,8 @@ class SQLInterpreterSpec extends BaseInterpreterSpec {
 
   override def createInterpreter(): Interpreter = {
     val conf = new SparkConf()
+      .set(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+      .set(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
     if (sparkEntries == null) {
       sparkEntries = new SparkEntries(conf)
     }

repl/src/test/scala/org/apache/livy/repl/ScalaInterpreterSpec.scala

@@ -21,6 +21,7 @@ import org.apache.spark.SparkConf
 import org.json4s.{DefaultFormats, JValue}
 import org.json4s.JsonDSL._
 
+import org.apache.livy.client.common.TestUtils
 import org.apache.livy.rsc.RSCConf
 
 class ScalaInterpreterSpec extends BaseInterpreterSpec {
@@ -32,7 +33,9 @@ class ScalaInterpreterSpec extends BaseInterpreterSpec {
   // against both scala-2.12 and scala-2.13 builds.
 
   override def createInterpreter(): Interpreter =
-    new SparkInterpreter(new SparkConf())
+    new SparkInterpreter(new SparkConf()
+      .set(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+      .set(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST))
 
   it should "execute `1 + 2` == 3" in withInterpreter { interpreter =>
     val response = interpreter.execute("1 + 2")

repl/src/test/scala/org/apache/livy/repl/SessionSpec.scala

@@ -28,6 +28,7 @@ import org.scalatest.matchers.should.Matchers._
 import org.scalatest.time._
 
 import org.apache.livy.LivyBaseUnitTestSuite
+import org.apache.livy.client.common.TestUtils
 import org.apache.livy.repl.Interpreter.ExecuteResponse
 import org.apache.livy.rsc.RSCConf
 import org.apache.livy.sessions._
@@ -39,6 +40,10 @@ class SessionSpec extends AnyFunSpec with Eventually
 
   private val rscConf = new RSCConf(new Properties()).set(RSCConf.Entry.SESSION_KIND, "spark")
 
+  private def newSparkConf(): SparkConf = new SparkConf()
+    .set(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+    .set(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
+
   describe("Session") {
     var session: Session = null
 
@@ -54,7 +59,7 @@ class SessionSpec extends AnyFunSpec with Eventually
         Array("not_started", "starting", "idle", "busy", "idle", "busy", "idle")
       val actualStateTransitions = new ConcurrentLinkedQueue[String]()
 
-      session = new Session(rscConf, new SparkConf(), None,
+      session = new Session(rscConf, newSparkConf(), None,
         { s => actualStateTransitions.add(s.toString) })
       session.start()
       session.execute("")
@@ -70,13 +75,13 @@ class SessionSpec extends AnyFunSpec with Eventually
       val actualStateTransitions = new ConcurrentLinkedQueue[String]()
 
       val blockFirstExecuteCall = new CountDownLatch(1)
-      val interpreter = new SparkInterpreter(new SparkConf()) {
+      val interpreter = new SparkInterpreter(newSparkConf()) {
         override def execute(code: String): ExecuteResponse = {
           blockFirstExecuteCall.await(10, TimeUnit.SECONDS)
           super.execute(code)
         }
       }
-      session = new Session(rscConf, new SparkConf(), Some(interpreter),
+      session = new Session(rscConf, newSparkConf(), Some(interpreter),
         { s => actualStateTransitions.add(s.toString) })
       session.start()
 
@@ -92,7 +97,7 @@ class SessionSpec extends AnyFunSpec with Eventually
 
     it("should remove old statements when reaching threshold") {
       rscConf.set(RSCConf.Entry.RETAINED_STATEMENTS, 2)
-      session = new Session(rscConf, new SparkConf())
+      session = new Session(rscConf, newSparkConf())
       session.start()
 
       session.statements.size should be (0)

repl/src/test/scala/org/apache/livy/repl/SparkRInterpreterSpec.scala

@@ -24,6 +24,7 @@ import org.scalatest.{BeforeAndAfterAll, Outcome}
 import org.scalatest.funspec.AnyFunSpec
 import org.scalatest.matchers.should.Matchers
 
+import org.apache.livy.client.common.TestUtils
 import org.apache.livy.rsc.driver.SparkEntries
 
 class SparkRInterpreterSpec extends BaseInterpreterSpec {
@@ -38,6 +39,8 @@ class SparkRInterpreterSpec extends BaseInterpreterSpec {
 
   override def createInterpreter(): Interpreter = {
     val sparkConf = new SparkConf()
+      .set(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+      .set(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
     SparkRInterpreter(sparkConf, new SparkEntries(sparkConf))
   }
 

rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java

@@ -82,6 +82,13 @@ private Properties createConf(boolean local, boolean hiveSupport) {
     conf.put("spark.sql.catalogImplementation", hiveSupport ? "hive" : "in-memory");
     conf.put(RETAINED_SHARE_VARIABLES.key(), "2");
     conf.put(RPC_SERVER_ADDRESS.key(), TestUtils.TEST_BIND_HOST);
+    // Pin Spark's own driver binding to loopback so this test passes on macOS
+    // without requiring an externally-exported SPARK_LOCAL_IP=127.0.0.1. On
+    // macOS InetAddress.getLocalHost resolves to the LAN IP, which a peer
+    // process on the same host cannot reach, causing connect timeouts.
+    // Harmless on Linux CI where loopback is equally reachable.
+    conf.put(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST);
+    conf.put(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST);
     return conf;
   }
 

scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTest.scala

@@ -191,6 +191,8 @@ object ScalaClientTest {
     conf.put(CLIENT_SHUTDOWN_TIMEOUT.key(), "30s")
     conf.put(LIVY_JARS.key, "")
     conf.put(RPC_SERVER_ADDRESS.key(), TestUtils.TEST_BIND_HOST)
+    conf.put(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+    conf.put(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
     conf
   }
 

server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala

@@ -67,6 +67,8 @@ abstract class BaseInteractiveServletSpec
       RSCConf.Entry.LIVY_JARS.key() -> "",
       RSCConf.Entry.CLIENT_IN_PROCESS.key() -> inProcess.toString,
       RSCConf.Entry.RPC_SERVER_ADDRESS.key() -> TestUtils.TEST_BIND_HOST,
+      TestUtils.SPARK_DRIVER_HOST -> TestUtils.TEST_BIND_HOST,
+      TestUtils.SPARK_DRIVER_BIND_ADDRESS -> TestUtils.TEST_BIND_HOST,
       SparkLauncher.SPARK_MASTER -> "local",
       SparkLauncher.DRIVER_EXTRA_CLASSPATH -> classpath,
       SparkLauncher.EXECUTOR_EXTRA_CLASSPATH -> classpath

server/src/test/scala/org/apache/livy/server/interactive/InteractiveSessionSpec.scala

@@ -72,7 +72,9 @@ class InteractiveSessionSpec extends AnyFunSpec
     req.conf = Map(
       SparkLauncher.DRIVER_EXTRA_CLASSPATH -> sys.props("java.class.path"),
       RSCConf.Entry.LIVY_JARS.key() -> "",
-      RSCConf.Entry.RPC_SERVER_ADDRESS.key() -> TestUtils.TEST_BIND_HOST
+      RSCConf.Entry.RPC_SERVER_ADDRESS.key() -> TestUtils.TEST_BIND_HOST,
+      TestUtils.SPARK_DRIVER_HOST -> TestUtils.TEST_BIND_HOST,
+      TestUtils.SPARK_DRIVER_BIND_ADDRESS -> TestUtils.TEST_BIND_HOST
     )
     InteractiveSession.create(0, None, null, None, livyConf, accessManager, req,
       sessionStore, None, None, mockApp)

thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerBaseTest.scala

@@ -48,11 +48,21 @@ abstract class ThriftServerBaseTest extends AnyFunSuite with BeforeAndAfterAll {
     formatSparkVersion(sparkVersion)
   }
 
-  def jdbcUri(defaultDb: String, sessionConf: String*): String = if (mode == ServerMode.http) {
-    s"jdbc:hive2://localhost:$port/$defaultDb?hive.server2.transport.mode=http;" +
-      s"hive.server2.thrift.http.path=cliservice;${sessionConf.mkString(";")}"
-  } else {
-    s"jdbc:hive2://localhost:$port/$defaultDb?${sessionConf.mkString(";")}"
+  def jdbcUri(defaultDb: String, sessionConf: String*): String = {
+    // Pin the Spark driver to loopback so that sessions launched by the Thrift server bind and
+    // advertise a locally reachable address. Without this, on hosts whose hostname resolves to a
+    // non-bindable LAN IP (e.g. some macOS setups) the Spark driver / block manager fails with
+    // "Can't assign requested address" or the executor times out fetching REPL classes.
+    val driverConf = Seq(
+      s"livy.session.conf.${TestUtils.SPARK_DRIVER_HOST}=${TestUtils.TEST_BIND_HOST}",
+      s"livy.session.conf.${TestUtils.SPARK_DRIVER_BIND_ADDRESS}=${TestUtils.TEST_BIND_HOST}")
+    val allConf = driverConf ++ sessionConf
+    if (mode == ServerMode.http) {
+      s"jdbc:hive2://localhost:$port/$defaultDb?hive.server2.transport.mode=http;" +
+        s"hive.server2.thrift.http.path=cliservice;${allConf.mkString(";")}"
+    } else {
+      s"jdbc:hive2://localhost:$port/$defaultDb?${allConf.mkString(";")}"
+    }
   }
 
   override def beforeAll(): Unit = {

thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ColumnBufferTest.java

@@ -40,6 +40,8 @@
 import org.junit.Test;
 import static org.junit.Assert.*;
 
+import org.apache.livy.client.common.TestUtils;
+
 public class ColumnBufferTest {
 
   @Test
@@ -50,6 +52,8 @@ public void testColumnBuffer() throws Exception {
       .master("local")
       .appName(getClass().getName())
       .config("spark.sql.warehouse.dir", warehouse)
+      .config(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST)
+      .config(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST)
       .getOrCreate();
 
     try {

thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ThriftSessionTest.java

@@ -55,6 +55,8 @@ public static void setUp() throws Exception {
     conf.put("spark.sql.warehouse.dir", warehouse);
     conf.put("spark.sql.catalogImplementation", "in-memory");
     conf.put(RPC_SERVER_ADDRESS.key(), TestUtils.TEST_BIND_HOST);
+    conf.put(TestUtils.SPARK_DRIVER_HOST, TestUtils.TEST_BIND_HOST);
+    conf.put(TestUtils.SPARK_DRIVER_BIND_ADDRESS, TestUtils.TEST_BIND_HOST);
 
     livy = new LivyClientBuilder(false)
       .setURI(new URI("rsc:/"))