[SPARK-52193][K8S] Add Spark Connect port to Spark Driver pod and service

### What changes were proposed in this pull request?

This PR aims to add `Spark Connect` port to `Spark Driver` pod and service.

### Why are the changes needed?

To support `Spark Connect` easily.

### Does this PR introduce _any_ user-facing change?

No behavior change because this is a new port.

### How was this patch tested?

Pass the CIs.

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

No.

Closes #50925 from dongjoon-hyun/SPARK-52193.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala

@@ -58,9 +58,11 @@ object Constants {
   // Default and fixed ports
   val DEFAULT_DRIVER_PORT = 7078
   val DEFAULT_BLOCKMANAGER_PORT = 7079
+  val DEFAULT_SPARK_CONNECT_SERVER_PORT = 15002
   val DRIVER_PORT_NAME = "driver-rpc-port"
   val BLOCK_MANAGER_PORT_NAME = "blockmanager"
   val UI_PORT_NAME = "spark-ui"
+  val SPARK_CONNECT_SERVER_PORT_NAME = "spark-connect"
 
   // Environment Variables
   val ENV_DRIVER_POD_IP = "SPARK_DRIVER_POD_IP"
@@ -104,6 +106,7 @@ object Constants {
   val DEFAULT_DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
   val DEFAULT_EXECUTOR_CONTAINER_NAME = "spark-kubernetes-executor"
   val NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.4d
+  val CONNECT_GRPC_BINDING_PORT = "spark.connect.grpc.binding.port"
 
   // Hadoop Configuration
   val HADOOP_CONF_VOLUME = "hadoop-properties"

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStep.scala

@@ -99,6 +99,8 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
       conf.sparkConf.getInt(BLOCK_MANAGER_PORT.key, DEFAULT_BLOCKMANAGER_PORT)
     )
     val driverUIPort = SparkUI.getUIPort(conf.sparkConf)
+    val driverSparkConnectServerPort =
+      conf.sparkConf.getInt(CONNECT_GRPC_BINDING_PORT, DEFAULT_SPARK_CONNECT_SERVER_PORT)
     val driverContainer = new ContainerBuilder(pod.container)
       .withName(Option(pod.container.getName).getOrElse(DEFAULT_DRIVER_CONTAINER_NAME))
       .withImage(driverContainerImage)
@@ -118,6 +120,11 @@ private[spark] class BasicDriverFeatureStep(conf: KubernetesDriverConf)
         .withContainerPort(driverUIPort)
         .withProtocol("TCP")
         .endPort()
+      .addNewPort()
+        .withName(SPARK_CONNECT_SERVER_PORT_NAME)
+        .withContainerPort(driverSparkConnectServerPort)
+        .withProtocol("TCP")
+        .endPort()
       .addNewEnv()
         .withName(ENV_SPARK_USER)
         .withValue(Utils.getCurrentUserName())

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStep.scala

@@ -48,6 +48,8 @@ private[spark] class DriverServiceFeatureStep(
   private val driverBlockManagerPort = kubernetesConf.sparkConf.getInt(
     config.DRIVER_BLOCK_MANAGER_PORT.key, DEFAULT_BLOCKMANAGER_PORT)
   private val  driverUIPort = kubernetesConf.get(config.UI.UI_PORT)
+  private val driverSparkConnectServerPort = kubernetesConf.sparkConf.getInt(
+    CONNECT_GRPC_BINDING_PORT, DEFAULT_SPARK_CONNECT_SERVER_PORT)
 
   override def configurePod(pod: SparkPod): SparkPod = pod
 
@@ -86,6 +88,11 @@ private[spark] class DriverServiceFeatureStep(
           .withPort(driverUIPort)
           .withNewTargetPort(driverUIPort)
           .endPort()
+        .addNewPort()
+          .withName(SPARK_CONNECT_SERVER_PORT_NAME)
+          .withPort(driverSparkConnectServerPort)
+          .withNewTargetPort(driverSparkConnectServerPort)
+          .endPort()
         .endSpec()
       .build()
     Seq(driverService)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicDriverFeatureStepSuite.scala

@@ -84,7 +84,8 @@ class BasicDriverFeatureStepSuite extends SparkFunSuite {
     val expectedPortNames = Set(
       containerPort(DRIVER_PORT_NAME, DEFAULT_DRIVER_PORT),
       containerPort(BLOCK_MANAGER_PORT_NAME, DEFAULT_BLOCKMANAGER_PORT),
-      containerPort(UI_PORT_NAME, UI_PORT.defaultValue.get)
+      containerPort(UI_PORT_NAME, UI_PORT.defaultValue.get),
+      containerPort(SPARK_CONNECT_SERVER_PORT_NAME, DEFAULT_SPARK_CONNECT_SERVER_PORT)
     )
     val foundPortNames = configuredPod.container.getPorts.asScala.toSet
     assert(expectedPortNames === foundPortNames)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStepSuite.scala

@@ -49,6 +49,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
       .set(DRIVER_PORT, 9000)
       .set(DRIVER_BLOCK_MANAGER_PORT, 8080)
       .set(UI_PORT, 4080)
+      .set(CONNECT_GRPC_BINDING_PORT, "15003")
     val kconf = KubernetesTestConf.createDriverConf(
       sparkConf = sparkConf,
       labels = DRIVER_LABELS,
@@ -66,6 +67,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
       9000,
       8080,
       4080,
+      15003,
       s"${kconf.resourceNamePrefix}${DriverServiceFeatureStep.DRIVER_SVC_POSTFIX}",
       kconf.appId,
       driverService)
@@ -100,6 +102,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
       DEFAULT_DRIVER_PORT,
       DEFAULT_BLOCKMANAGER_PORT,
       UI_PORT.defaultValue.get,
+      DEFAULT_SPARK_CONNECT_SERVER_PORT,
       s"${kconf.resourceNamePrefix}${DriverServiceFeatureStep.DRIVER_SVC_POSTFIX}",
       kconf.appId,
       resolvedService)
@@ -233,6 +236,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
       driverPort: Int,
       blockManagerPort: Int,
       drierUIPort: Int,
+      driverConnectServerPort: Int,
       expectedServiceName: String,
       appId: String,
       service: Service): Unit = {
@@ -249,7 +253,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
     DRIVER_SERVICE_ANNOTATIONS.foreach { case (k, v) =>
       assert(service.getMetadata.getAnnotations.get(k) === v)
     }
-    assert(service.getSpec.getPorts.size() === 3)
+    assert(service.getSpec.getPorts.size() === 4)
     val driverServicePorts = service.getSpec.getPorts.asScala
     assert(driverServicePorts.head.getName === DRIVER_PORT_NAME)
     assert(driverServicePorts.head.getPort.intValue() === driverPort)
@@ -260,5 +264,7 @@ class DriverServiceFeatureStepSuite extends SparkFunSuite {
     assert(driverServicePorts(2).getName === UI_PORT_NAME)
     assert(driverServicePorts(2).getPort.intValue() === drierUIPort)
     assert(driverServicePorts(2).getTargetPort.getIntVal === drierUIPort)
+    assert(driverServicePorts(3).getPort.intValue() === driverConnectServerPort)
+    assert(driverServicePorts(3).getTargetPort.getIntVal === driverConnectServerPort)
   }
 }