Require block manager port to be configured

Author: EnricoMi

common/utils/src/main/resources/error/error-conditions.json

@@ -1499,6 +1499,12 @@
     ],
     "sqlState" : "42702"
   },
+  "EXECUTOR_KUBERNETES_SERVICE_REQUIRES_BLOCK_MANAGER_PORT" : {
+    "message" : [
+      "Enabling the executor Kubernetes service requires <blockManagerPortConfigKey> to be set to a positive number, for instance <defaultShuffleServicePort>."
+    ],
+    "sqlState" : "42000"
+  },
   "EXEC_IMMEDIATE_DUPLICATE_ARGUMENT_ALIASES" : {
     "message" : [
       "The USING clause of this EXECUTE IMMEDIATE command contained multiple arguments with same alias (<aliases>), which is invalid; please update the command to specify unique aliases and then try it again."

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

@@ -419,12 +419,13 @@ private[spark] object Config extends Logging {
   val KUBERNETES_EXECUTOR_ENABLE_SERVICE =
     ConfigBuilder("spark.kubernetes.executor.enableService")
       .doc("If a Kubernetes service is created for the executor. " +
-        "The executor pod creates a Kubernetes service that allows to connect to executor ports " +
-        "via the Kubernetes service instead of the pod host IP. Connecting to such ports " +
-        "instantly fails with 'connection refused' error once the executor got decommissioned. " +
+        "A Kubernetes service is created for the executor pod that allows to connect to executor " +
+        "ports via the Kubernetes service instead of the pod host IP. Once the executor got " +
+        "decommissioned, connecting to such ports instantly fails with 'connection refused'. " +
         "Connection to the port via the pod host IP instead fails with a 'connection timeout' " +
-        "which is set via NETWORK_TIMEOUT and defaults to 2 minutes. " +
-        "The kubernetes service provides access to the executors shuffle service.")
+        "after NETWORK_TIMEOUT, which defaults to 2 minutes. " +
+        "The executor kubernetes service provides access to the executor's block manager, so " +
+        "BLOCK_MANAGER_PORT has to be given a value greater than zero.")
       .version("4.1.0")
       .booleanConf
       .createWithDefault(false)

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

@@ -22,19 +22,30 @@ import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, ServiceBu
 
 import org.apache.spark.SparkException
 import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, SparkPod}
-import org.apache.spark.internal.config.SHUFFLE_SERVICE_PORT
+import org.apache.spark.internal.config.{BLOCK_MANAGER_PORT, SHUFFLE_SERVICE_PORT}
 
 class ExecutorServiceFeatureStep(conf: KubernetesExecutorConf) extends KubernetesFeatureConfigStep {
   private val spark_app_selector_label = "spark-app-selector"
   private val spark_exec_id_label = "spark-exec-id"
   private val service_selector_labels = Set(spark_app_selector_label, spark_exec_id_label)
+  private lazy val selector = conf.labels
+    .filter { case (key, _) => service_selector_labels.contains(key) }
 
   private lazy val sparkAppSelector = getLabel(spark_app_selector_label)
   private lazy val sparkExecId = getLabel(spark_exec_id_label)
   // name length is 8 + 38 + 6 + 10 = 62
   // which fits in KUBERNETES_DNS_LABEL_NAME_MAX_LENGTH = 63
   private lazy val serviceName = s"svc-$sparkAppSelector-exec-$sparkExecId"
 
+  // The executor kubernetes services requires BLOCK_MANAGER_PORT to be set
+  private val blockManagerPortName = "spark-block-manager"
+  private val blockManagerPort = conf.sparkConf.get(BLOCK_MANAGER_PORT)
+  SparkException.require(blockManagerPort > 0,
+    "EXECUTOR_KUBERNETES_SERVICE_REQUIRES_BLOCK_MANAGER_PORT",
+    Map(
+      "blockManagerPortConfigKey" -> BLOCK_MANAGER_PORT.key,
+      "defaultShuffleServicePort" -> SHUFFLE_SERVICE_PORT.defaultValue.get.toString));
+
   private def getLabel(label: String): String = {
     val value = conf.labels.get(label)
     value.getOrElse(
@@ -55,23 +66,16 @@ class ExecutorServiceFeatureStep(conf: KubernetesExecutorConf) extends Kubernete
   }
 
   override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
-    val selector = conf.labels
-      .filter { case (key, _) => service_selector_labels.contains(key) }
-
-    // kubernetes executor service provides access to the executor's shuffle service
-    val portName = "spark-shuffle-service"
-    val port = conf.sparkConf.get(SHUFFLE_SERVICE_PORT)
-
     val service = new ServiceBuilder()
       .withNewMetadata()
       .withName(serviceName)
       .endMetadata()
       .withNewSpec()
       .withSelector(selector.asJava)
       .addNewPort()
-      .withName(portName)
-      .withPort(port)
-      .withNewTargetPort(port)
+      .withName(blockManagerPortName)
+      .withPort(blockManagerPort)
+      .withNewTargetPort(blockManagerPort)
       .endPort()
       .endSpec()
       .build()

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala

@@ -22,10 +22,10 @@ import io.fabric8.kubernetes.api.model.Service
 import io.fabric8.kubernetes.client.KubernetesClient
 import org.mockito.Mockito.mock
 
-import org.apache.spark.{SecurityManager, SparkConf}
+import org.apache.spark.{SecurityManager, SparkConf, SparkIllegalArgumentException}
 import org.apache.spark.deploy.k8s._
 import org.apache.spark.deploy.k8s.features.KubernetesExecutorCustomFeatureConfigStep
-import org.apache.spark.internal.config.{ConfigEntry, SHUFFLE_SERVICE_PORT}
+import org.apache.spark.internal.config.{BLOCK_MANAGER_PORT, ConfigEntry}
 import org.apache.spark.resource.ResourceProfile
 
 class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
@@ -81,29 +81,39 @@ class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
   }
 
   test("SPARK-XXXXX: check executor kubernetes spec with service enabled") {
-    Seq(None, Some(1234)).foreach { somePort =>
-      val sparkConf = baseConf.clone.set(Config.KUBERNETES_EXECUTOR_ENABLE_SERVICE, true)
-      somePort.foreach(sparkConf.set(SHUFFLE_SERVICE_PORT, _))
-      val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
-      val secMgr = new SecurityManager(sparkConf)
-      val client = mock(classOf[KubernetesClient])
-      val profile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
-      val spec = new KubernetesExecutorBuilder().buildFromFeatures(conf, secMgr, client, profile)
-
-      val containerEnvs = spec.pod.container.getEnv.asScala
-      assert(containerEnvs.exists(_.getName === "EXECUTOR_SERVICE_NAME"))
-      val containerEnv = containerEnvs.filter(_.getName === "EXECUTOR_SERVICE_NAME").head
-      assert(containerEnv.getValue === "svc-appId-exec-1")
-
-      assert(spec.executorKubernetesResources.size === 1)
-      val resource = spec.executorKubernetesResources.head
-      assert(resource.getKind === "Service")
-      val service = resource.asInstanceOf[Service]
-      assert(service.getMetadata.getName === "svc-appId-exec-1")
-      assert(service.getSpec.getPorts.size() === 1)
-      val port = service.getSpec.getPorts.get(0)
-      assert(port.getName === "spark-shuffle-service")
-      assert(port.getPort === somePort.getOrElse(SHUFFLE_SERVICE_PORT.defaultValue.get))
+    val sparkConf = baseConf.clone
+      .set(Config.KUBERNETES_EXECUTOR_ENABLE_SERVICE, true)
+      .set(BLOCK_MANAGER_PORT, 1234)
+    val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
+    val secMgr = new SecurityManager(sparkConf)
+    val client = mock(classOf[KubernetesClient])
+    val profile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
+    val spec = new KubernetesExecutorBuilder().buildFromFeatures(conf, secMgr, client, profile)
+
+    val containerEnvs = spec.pod.container.getEnv.asScala
+    assert(containerEnvs.exists(_.getName === "EXECUTOR_SERVICE_NAME"))
+    val containerEnv = containerEnvs.filter(_.getName === "EXECUTOR_SERVICE_NAME").head
+    assert(containerEnv.getValue === "svc-appId-exec-1")
+
+    assert(spec.executorKubernetesResources.size === 1)
+    val resource = spec.executorKubernetesResources.head
+    assert(resource.getKind === "Service")
+    val service = resource.asInstanceOf[Service]
+    assert(service.getMetadata.getName === "svc-appId-exec-1")
+    assert(service.getSpec.getPorts.size() === 1)
+    val port = service.getSpec.getPorts.get(0)
+    assert(port.getName === "spark-block-manager")
+    assert(port.getPort === 1234)
+  }
+
+  test("SPARK-XXXXX: check executor kubernetes service requires block manager port") {
+    val sparkConf = baseConf.clone.set(Config.KUBERNETES_EXECUTOR_ENABLE_SERVICE, true)
+    val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
+    val secMgr = new SecurityManager(sparkConf)
+    val client = mock(classOf[KubernetesClient])
+    val profile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
+    assertThrows[SparkIllegalArgumentException] {
+      new KubernetesExecutorBuilder().buildFromFeatures(conf, secMgr, client, profile)
     }
   }
 }