Allow minExecutors=0 in dynamic mode (#89)

Closes G-Research/spark#157

---------

Signed-off-by: Sudipto Baral <sudiptobaral.me@gmail.com>

Author: sudiptob2

src/main/scala/org/apache/spark/deploy/armada/submit/ArmadaClientApplication.scala

@@ -81,7 +81,7 @@ import org.apache.spark.deploy.k8s.Config.{
   KUBERNETES_SUBMIT_GRACE_PERIOD
 }
 import org.apache.spark.{SecurityManager, SparkConf}
-import org.apache.spark.internal.config.{DRIVER_PORT, DRIVER_HOST_ADDRESS}
+import org.apache.spark.internal.config.{DRIVER_PORT, DRIVER_HOST_ADDRESS, DYN_ALLOCATION_ENABLED}
 import org.apache.spark.resource.ResourceProfile
 import org.apache.spark.scheduler.cluster.k8s.KubernetesExecutorBuilder
 import io.fabric8.kubernetes.client.DefaultKubernetesClient
@@ -363,12 +363,6 @@ private[spark] class ArmadaClientApplication extends SparkApplication {
       driverJobId: String,
       executorCount: Int
   ): Seq[String] = {
-    if (executorCount <= 0) {
-      throw new IllegalArgumentException(
-        s"Executor count must be greater than 0, but got: $executorCount"
-      )
-    }
-
     val driverData = buildDriverData(None, armadaJobConfig, conf)
 
     val executorLabels = buildLabels(
@@ -443,9 +437,12 @@ private[spark] class ArmadaClientApplication extends SparkApplication {
   ): (String, Seq[String]) = {
     val modeHelper    = DeploymentModeHelper(conf)
     val executorCount = modeHelper.getExecutorCount
-    if (executorCount <= 0) {
+    val isDynamic     = conf.getBoolean(DYN_ALLOCATION_ENABLED.key, false)
+
+    // Allow minExecutors=0 for dynamic allocation
+    if (!isDynamic && executorCount <= 0) {
       throw new IllegalArgumentException(
-        s"Executor count must be greater than 0, but got: $executorCount"
+        s"Executor count must be greater than 0 for static allocation, but got: $executorCount"
       )
     }
 
@@ -1783,12 +1780,14 @@ private[spark] class ArmadaClientApplication extends SparkApplication {
       nodeUniformityLabel: Option[String],
       conf: SparkConf
   ): Map[String, String] = {
-    val modeHelper = DeploymentModeHelper(conf)
+    val modeHelper      = DeploymentModeHelper(conf)
+    val gangCardinality = modeHelper.getGangCardinality
     configGenerator.getAnnotations ++ templateAnnotations ++ nodeUniformityLabel
+      .filter(_ => gangCardinality > 0) // Only add gang annotations if cardinality > 0
       .map(label =>
         GangSchedulingAnnotations(
           gangId,
-          modeHelper.getGangCardinality,
+          gangCardinality,
           label
         )
       )

src/test/scala/org/apache/spark/deploy/armada/submit/ArmadaClientApplicationSuite.scala

@@ -1653,7 +1653,9 @@ class ArmadaClientApplicationSuite extends AnyFunSuite with BeforeAndAfter with
     }
   }
 
-  test("submitArmadaJob should validate executor count is greater than zero") {
+  test(
+    "submitArmadaJob should validate executor count is greater than zero for static allocation"
+  ) {
     val armadaJobConfig = armadaClientApp.ArmadaJobConfig(
       queue = "test-queue",
       jobSetId = "test-job-set",
@@ -1685,13 +1687,58 @@ class ArmadaClientApplicationSuite extends AnyFunSuite with BeforeAndAfter with
     )
 
     sparkConf.set("spark.executor.instances", "0")
+    sparkConf.set("spark.dynamicAllocation.enabled", "false")
 
     val exception = intercept[IllegalArgumentException] {
       armadaClientApp.submitArmadaJob(null, clientArguments, armadaJobConfig, sparkConf)
     }
     exception.getMessage should include("Executor count must be greater than 0")
   }
 
+  test("submitArmadaJob should allow minExecutors=0 for dynamic allocation") {
+    val armadaJobConfig = armadaClientApp.ArmadaJobConfig(
+      queue = "test-queue",
+      jobSetId = "test-job-set",
+      jobTemplate = None,
+      driverJobItemTemplate = None,
+      executorJobItemTemplate = None,
+      cliConfig = armadaClientApp.CLIConfig(
+        queue = Some("test-queue"),
+        jobSetId = Some("test-job-set"),
+        namespace = Some("test-namespace"),
+        priority = Some(RUNTIME_PRIORITY),
+        containerImage = Some(DEFAULT_IMAGE_NAME),
+        podLabels = Map.empty,
+        driverLabels = Map.empty,
+        executorLabels = Map.empty,
+        armadaClusterUrl = Some("armada://localhost:50051"),
+        nodeSelectors = Map.empty,
+        nodeUniformityLabel = None,
+        executorConnectionTimeout = Some(300.seconds),
+        runAsUser = None,
+        driverResources = armadaClientApp.ResourceConfig(None, None, None, None),
+        executorResources = armadaClientApp.ResourceConfig(None, None, None, None)
+      ),
+      applicationId = "armada-spark-app-id",
+      driverFeatureStepJobItem = None,
+      driverFeatureStepContainer = None,
+      executorFeatureStepJobItem = None,
+      executorFeatureStepContainer = None
+    )
+
+    sparkConf.set("spark.submit.deployMode", "cluster")
+    sparkConf.set("spark.dynamicAllocation.enabled", "true")
+    sparkConf.set("spark.dynamicAllocation.minExecutors", "0")
+
+    // Should not throw executor count validation exception
+    val exception = intercept[Throwable] {
+      armadaClientApp.submitArmadaJob(null, clientArguments, armadaJobConfig, sparkConf)
+    }
+    Option(exception.getMessage).getOrElse(
+      ""
+    ) should not include "Executor count must be greater than 0"
+  }
+
   test("JobTemplateLoader should handle malformed YAML gracefully") {
     val malformedYaml = "invalid: yaml: content: ]]]["
     val templateFile  = tempDir.resolve("malformed-template.yaml").toFile