diff --git a/client-common/src/main/java/org/apache/livy/client/common/TestUtils.java b/client-common/src/main/java/org/apache/livy/client/common/TestUtils.java index 6689cf565..24cfb6ff8 100644 --- a/client-common/src/main/java/org/apache/livy/client/common/TestUtils.java +++ b/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" diff --git a/integration-test/src/main/scala/org/apache/livy/test/framework/MiniCluster.scala b/integration-test/src/main/scala/org/apache/livy/test/framework/MiniCluster.scala index d66b1d620..48428eafe 100644 --- a/integration-test/src/main/scala/org/apache/livy/test/framework/MiniCluster.scala +++ b/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 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 e8a80e372..84fb5559b 100644 --- a/repl/src/test/scala/org/apache/livy/repl/BaseSessionSpec.scala +++ b/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) diff --git a/repl/src/test/scala/org/apache/livy/repl/PythonInterpreterSpec.scala b/repl/src/test/scala/org/apache/livy/repl/PythonInterpreterSpec.scala index 61e27ad6f..e47a6f04a 100644 --- a/repl/src/test/scala/org/apache/livy/repl/PythonInterpreterSpec.scala +++ b/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)) } diff --git a/repl/src/test/scala/org/apache/livy/repl/ReplDriverSuite.scala b/repl/src/test/scala/org/apache/livy/repl/ReplDriverSuite.scala index 66947d6af..4b4f0cb62 100644 --- a/repl/src/test/scala/org/apache/livy/repl/ReplDriverSuite.scala +++ b/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) diff --git a/repl/src/test/scala/org/apache/livy/repl/SQLInterpreterSpec.scala b/repl/src/test/scala/org/apache/livy/repl/SQLInterpreterSpec.scala index 5e839d4c8..37a10c341 100644 --- a/repl/src/test/scala/org/apache/livy/repl/SQLInterpreterSpec.scala +++ b/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) } 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 25b62668e..b0a23afd5 100644 --- a/repl/src/test/scala/org/apache/livy/repl/ScalaInterpreterSpec.scala +++ b/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") diff --git a/repl/src/test/scala/org/apache/livy/repl/SessionSpec.scala b/repl/src/test/scala/org/apache/livy/repl/SessionSpec.scala index af47e0b63..ba5433e52 100644 --- a/repl/src/test/scala/org/apache/livy/repl/SessionSpec.scala +++ b/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) diff --git a/repl/src/test/scala/org/apache/livy/repl/SparkRInterpreterSpec.scala b/repl/src/test/scala/org/apache/livy/repl/SparkRInterpreterSpec.scala index dd69a6869..6f39a340b 100644 --- a/repl/src/test/scala/org/apache/livy/repl/SparkRInterpreterSpec.scala +++ b/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)) } 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 44bfa654e..98886a037 100644 --- a/rsc/src/test/java/org/apache/livy/rsc/TestSparkClient.java +++ b/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; } diff --git a/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTest.scala b/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTest.scala index fb751c063..fb335b348 100644 --- a/scala-api/src/test/scala/org/apache/livy/scalaapi/ScalaClientTest.scala +++ b/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 } diff --git a/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala b/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala index 9803787f7..9c4ef91f3 100644 --- a/server/src/test/scala/org/apache/livy/server/interactive/BaseInteractiveServletSpec.scala +++ b/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 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 1efa266ac..b7a633ac1 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 @@ -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) diff --git a/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerBaseTest.scala b/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerBaseTest.scala index 860df01e9..0cdf664c0 100644 --- a/thriftserver/server/src/test/scala/org/apache/livy/thriftserver/ThriftServerBaseTest.scala +++ b/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 = { diff --git a/thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ColumnBufferTest.java b/thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ColumnBufferTest.java index b8832fb07..4c1e2e3e5 100644 --- a/thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ColumnBufferTest.java +++ b/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 { diff --git a/thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ThriftSessionTest.java b/thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ThriftSessionTest.java index 4cecbdb2d..b571cdb86 100644 --- a/thriftserver/session/src/test/java/org/apache/livy/thriftserver/session/ThriftSessionTest.java +++ b/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:/"))