From ff854455da3110feb1dfe32fc1cfd9e58829bbed Mon Sep 17 00:00:00 2001 From: Matthew Ball Date: Wed, 22 Jul 2026 12:50:00 -0700 Subject: [PATCH 1/3] [SPARK-58264][CONNECT][TESTS] Use OS-assigned ports in Connect server tests --- .../sql/connect/test/RemoteSparkSession.scala | 10 ++-- .../sql/connect/SparkConnectServerTest.scala | 16 +++--- .../SparkConnectServiceKeepAliveSuite.scala | 54 +++++++------------ 3 files changed, 33 insertions(+), 47 deletions(-) diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala index 8bd6c5cf01681..a90bff1c5ff85 100644 --- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala +++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala @@ -18,6 +18,7 @@ package org.apache.spark.sql.connect.test import java.io.{File, IOException, OutputStream} import java.lang.ProcessBuilder.Redirect +import java.net.ServerSocket import java.nio.file.Paths import java.util.concurrent.TimeUnit @@ -33,7 +34,6 @@ import org.scalatest.time.SpanSugar._ // scalastyle:ignore import org.apache.spark.SparkBuildInfo import org.apache.spark.sql.connect.SparkSession import org.apache.spark.sql.connect.client.{RetryPolicy, SparkConnectClient} -import org.apache.spark.sql.connect.common.config.ConnectCommon import org.apache.spark.sql.connect.test.IntegrationTestUtils._ import org.apache.spark.util.ArrayImplicits._ @@ -52,9 +52,11 @@ object SparkConnectServerUtils { // The equivalent command to start the connect server via command line: // bin/spark-shell --conf spark.plugins=org.apache.spark.sql.connect.SparkConnectPlugin - // Server port - val port: Int = - ConnectCommon.CONNECT_GRPC_BINDING_PORT + util.Random.nextInt(1000) + // Bind a throwaway socket to port 0 to obtain an OS-assigned free port for the server process. + val port: Int = { + val socket = new ServerSocket(0) + try socket.getLocalPort finally socket.close() + } @volatile private var stopped = false diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala index c935370f0643f..cf49b5d0fca25 100644 --- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala +++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala @@ -32,7 +32,6 @@ import org.apache.spark.sql.classic import org.apache.spark.sql.connect import org.apache.spark.sql.connect.client.{CustomSparkConnectBlockingStub, ExecutePlanResponseReattachableIterator, RetryPolicy, SparkConnectClient, SparkConnectStubState} import org.apache.spark.sql.connect.client.arrow.ArrowSerializer -import org.apache.spark.sql.connect.common.config.ConnectCommon import org.apache.spark.sql.connect.config.Connect import org.apache.spark.sql.connect.dsl.MockRemoteSession import org.apache.spark.sql.connect.dsl.plans._ @@ -46,9 +45,8 @@ import org.apache.spark.sql.util.CloseableIterator */ trait SparkConnectServerTest extends SharedSparkSession { - // Server port - val serverPort: Int = - ConnectCommon.CONNECT_GRPC_BINDING_PORT + util.Random.nextInt(1000) + // The port the running service is bound to, set after start. + var serverPort: Int = -1 val eventuallyTimeout = 30.seconds @@ -65,12 +63,16 @@ trait SparkConnectServerTest extends SharedSparkSession { // Other suites using mocks leave a mess in the global executionManager, // shut it down so that it's cleared before starting server. SparkConnectService.executionManager.shutdown() - // Start the real service. + startService() + } + + /** Starts the service on an OS-assigned free port (port 0) and records it in `serverPort`. */ + protected def startService(confs: Seq[(String, String)] = extraServerConfs): Unit = { withSparkEnvConfs( - (Seq( - (Connect.CONNECT_GRPC_BINDING_PORT.key, serverPort.toString)) ++ extraServerConfs): _*) { + (Seq((Connect.CONNECT_GRPC_BINDING_PORT.key, "0")) ++ confs): _*) { SparkConnectService.start(spark.sparkContext) } + serverPort = SparkConnectService.localPort } override def afterAll(): Unit = { diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala index de8d6b84e30d1..a0c5b1951ded5 100644 --- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala +++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala @@ -53,6 +53,12 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") + // Stops the service and restarts it on a fresh OS-assigned port with the given confs. + private def restartService(confs: Seq[(String, String)]): Unit = { + SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) + startService(confs) + } + test("SPARK-58094: real SparkConnectService applies configured keepalive end-to-end") { val serverSession = SparkConnectService @@ -121,14 +127,10 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { // behavior), this client's cadence would violate that coupled invariant; with today's fixed, // decoupled GRPC_KEEPALIVE_PERMIT_TIME_SECONDS floor (10s), it's comfortably tolerated. val clientKeepAliveMs = (SparkConnectService.GRPC_KEEPALIVE_PERMIT_TIME_SECONDS + 1) * 1000 - SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) - withSparkEnvConfs( - Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, + restartService(Seq( Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "20s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "5s") { - SparkConnectService.start(spark.sparkContext) - } + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "5s")) try { val client = SparkConnectClient .builder() @@ -156,14 +158,10 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { client.shutdown() } } finally { - SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) - withSparkEnvConfs( - Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, + restartService(Seq( Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { - SparkConnectService.start(spark.sparkContext) - } + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) } } @@ -190,14 +188,10 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { // truly idle connection -- hence one call to establish the transport, then real idle time // (no further calls) spanning several ping intervals, then a connection-count check. val clientKeepAliveMs = (SparkConnectService.GRPC_KEEPALIVE_PERMIT_TIME_SECONDS + 1) * 1000 - SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) - withSparkEnvConfs( - Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, + restartService(Seq( Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { - SparkConnectService.start(spark.sparkContext) - } + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) val relay = new FreezableTcpRelay(serverPort) try { val client = SparkConnectClient @@ -238,14 +232,10 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { } } finally { relay.close() - SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) - withSparkEnvConfs( - Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, + restartService(Seq( Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { - SparkConnectService.start(spark.sparkContext) - } + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) } } @@ -253,14 +243,10 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { "SPARK-58094: disabling spark.connect.grpc.keepAlive.enabled reverts to the pre-fix hang") { // Restart the real service with keepalive fully disabled (not just given short/aggressive // timing) to prove the flag genuinely gates the fix rather than only tuning its timing. - SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) - withSparkEnvConfs( - Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, + restartService(Seq( Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { - SparkConnectService.start(spark.sparkContext) - } + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) try { val serverSession = SparkConnectService @@ -321,14 +307,10 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { } } finally { // Restore the enabled server for afterAll()/subsequent tests in this suite. - SparkConnectService.stop(Some(30), Some(TimeUnit.SECONDS)) - withSparkEnvConfs( - Connect.CONNECT_GRPC_BINDING_PORT.key -> serverPort.toString, + restartService(Seq( Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s") { - SparkConnectService.start(spark.sparkContext) - } + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) } } } From 47c5931eb36ccfdaf191f70c9656da45a4c075c2 Mon Sep 17 00:00:00 2001 From: Matthew Ball Date: Mon, 27 Jul 2026 12:55:15 -0700 Subject: [PATCH 2/3] [SPARK-58264][CONNECT] lint --- .../sql/connect/test/RemoteSparkSession.scala | 3 +- .../sql/connect/SparkConnectServerTest.scala | 3 +- .../SparkConnectServiceKeepAliveSuite.scala | 54 ++++++++++--------- 3 files changed, 33 insertions(+), 27 deletions(-) diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala index a90bff1c5ff85..b3dcaa43f09d6 100644 --- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala +++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala @@ -55,7 +55,8 @@ object SparkConnectServerUtils { // Bind a throwaway socket to port 0 to obtain an OS-assigned free port for the server process. val port: Int = { val socket = new ServerSocket(0) - try socket.getLocalPort finally socket.close() + try socket.getLocalPort + finally socket.close() } @volatile private var stopped = false diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala index cf49b5d0fca25..68adb498fcce6 100644 --- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala +++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/SparkConnectServerTest.scala @@ -68,8 +68,7 @@ trait SparkConnectServerTest extends SharedSparkSession { /** Starts the service on an OS-assigned free port (port 0) and records it in `serverPort`. */ protected def startService(confs: Seq[(String, String)] = extraServerConfs): Unit = { - withSparkEnvConfs( - (Seq((Connect.CONNECT_GRPC_BINDING_PORT.key, "0")) ++ confs): _*) { + withSparkEnvConfs((Seq((Connect.CONNECT_GRPC_BINDING_PORT.key, "0")) ++ confs): _*) { SparkConnectService.start(spark.sparkContext) } serverPort = SparkConnectService.localPort diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala index a0c5b1951ded5..ff27b09c9206a 100644 --- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala +++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectServiceKeepAliveSuite.scala @@ -127,10 +127,11 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { // behavior), this client's cadence would violate that coupled invariant; with today's fixed, // decoupled GRPC_KEEPALIVE_PERMIT_TIME_SECONDS floor (10s), it's comfortably tolerated. val clientKeepAliveMs = (SparkConnectService.GRPC_KEEPALIVE_PERMIT_TIME_SECONDS + 1) * 1000 - restartService(Seq( - Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", - Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "20s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "5s")) + restartService( + Seq( + Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", + Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "20s", + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "5s")) try { val client = SparkConnectClient .builder() @@ -158,10 +159,11 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { client.shutdown() } } finally { - restartService(Seq( - Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", - Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) + restartService( + Seq( + Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", + Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) } } @@ -188,10 +190,11 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { // truly idle connection -- hence one call to establish the transport, then real idle time // (no further calls) spanning several ping intervals, then a connection-count check. val clientKeepAliveMs = (SparkConnectService.GRPC_KEEPALIVE_PERMIT_TIME_SECONDS + 1) * 1000 - restartService(Seq( - Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", - Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) + restartService( + Seq( + Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", + Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) val relay = new FreezableTcpRelay(serverPort) try { val client = SparkConnectClient @@ -232,10 +235,11 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { } } finally { relay.close() - restartService(Seq( - Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", - Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) + restartService( + Seq( + Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", + Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) } } @@ -243,10 +247,11 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { "SPARK-58094: disabling spark.connect.grpc.keepAlive.enabled reverts to the pre-fix hang") { // Restart the real service with keepalive fully disabled (not just given short/aggressive // timing) to prove the flag genuinely gates the fix rather than only tuning its timing. - restartService(Seq( - Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", - Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) + restartService( + Seq( + Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "false", + Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) try { val serverSession = SparkConnectService @@ -307,10 +312,11 @@ class SparkConnectServiceKeepAliveSuite extends SparkConnectServerTest { } } finally { // Restore the enabled server for afterAll()/subsequent tests in this suite. - restartService(Seq( - Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", - Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", - Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) + restartService( + Seq( + Connect.CONNECT_GRPC_KEEPALIVE_ENABLED.key -> "true", + Connect.CONNECT_GRPC_KEEPALIVE_TIME.key -> "1s", + Connect.CONNECT_GRPC_KEEPALIVE_TIMEOUT.key -> "1s")) } } } From 27b797dce25190883185ca1b4b65739838683d46 Mon Sep 17 00:00:00 2001 From: Matthew Ball Date: Mon, 27 Jul 2026 15:01:41 -0700 Subject: [PATCH 3/3] [SPARK-58264][CONNECT] Fix Time-of-Check to Time-of-Use bug that could cause issues in parallel processes --- .../sql/connect/test/RemoteSparkSession.scala | 27 +++++++++++++------ .../connect/SimpleSparkConnectService.scala | 9 +++++++ 2 files changed, 28 insertions(+), 8 deletions(-) diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala index b3dcaa43f09d6..386aeba9011bb 100644 --- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala +++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/test/RemoteSparkSession.scala @@ -18,8 +18,8 @@ package org.apache.spark.sql.connect.test import java.io.{File, IOException, OutputStream} import java.lang.ProcessBuilder.Redirect -import java.net.ServerSocket -import java.nio.file.Paths +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Paths} import java.util.concurrent.TimeUnit import scala.concurrent.duration.FiniteDuration @@ -52,11 +52,21 @@ object SparkConnectServerUtils { // The equivalent command to start the connect server via command line: // bin/spark-shell --conf spark.plugins=org.apache.spark.sql.connect.SparkConnectPlugin - // Bind a throwaway socket to port 0 to obtain an OS-assigned free port for the server process. - val port: Int = { - val socket = new ServerSocket(0) - try socket.getLocalPort - finally socket.close() + // File the server process writes its actual bound port into. + private val portFile: File = { + val f = File.createTempFile("spark-connect-server-port", ".tmp") + f.deleteOnExit() + f + } + + // The port the launched server bound to, read from `portFile` once the server reports it. + lazy val port: Int = { + start() + eventually(timeout(1.minute)) { + val reported = new String(Files.readAllBytes(portFile.toPath), UTF_8).trim + assert(reported.nonEmpty, "The Spark Connect server has not reported its port yet.") + reported.toInt + } } @volatile private var stopped = false @@ -78,7 +88,8 @@ object SparkConnectServerUtils { command += "--driver-class-path" += connectJar command += "--class" += "org.apache.spark.sql.connect.SimpleSparkConnectService" command += "--jars" += catalystTestJar - command += "--conf" += s"spark.connect.grpc.binding.port=$port" + command += "--conf" += "spark.connect.grpc.binding.port=0" + command += "--conf" += s"spark.connect.test.portFile=${portFile.getAbsolutePath}" command ++= testConfigs command ++= log4jConfigs command += connectJar diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala index 8061e913dc0da..06c8dbcc0f6a4 100644 --- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala +++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/SimpleSparkConnectService.scala @@ -17,6 +17,8 @@ package org.apache.spark.sql.connect +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Paths} import java.util.concurrent.TimeUnit import scala.io.StdIn @@ -38,6 +40,9 @@ import org.apache.spark.sql.internal.SQLConf private[sql] object SimpleSparkConnectService { private val stopCommand = "q" + // Conf naming a file to write the actual bound port into. + private val portFileConf = "spark.connect.test.portFile" + def main(args: Array[String]): Unit = { val conf = new SparkConf() .set("spark.plugins", "org.apache.spark.sql.connect.SparkConnectPlugin") @@ -45,6 +50,10 @@ private[sql] object SimpleSparkConnectService { .set(SQLConf.ARTIFACTS_SESSION_ISOLATION_ALWAYS_APPLY_CLASSLOADER, true) val sparkSession = SparkSession.builder().config(conf).getOrCreate() val sparkContext = sparkSession.sparkContext // init spark context + // Write the actual bound port to the file, if one was configured. + sparkContext.getConf.getOption(portFileConf).foreach { path => + Files.write(Paths.get(path), SparkConnectService.localPort.toString.getBytes(UTF_8)) + } // scalastyle:off println println("Ready for client connections.") // scalastyle:on println