diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOps.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOps.scala index bd8e0f97132dd..217b236af6733 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOps.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOps.scala @@ -50,7 +50,11 @@ private class KillApplication extends K8sSubmitOp { (implicit client: KubernetesClient): Unit = { val podToDelete = getPod(namespace, pName) - if (Option(podToDelete).isDefined) { + // `getPod` returns a request handle, which is never null; only resolving it reports whether + // the pod exists. Without the `get()` the check below is always true, so a name that is not + // in the cluster would issue a delete that the API server answers with a swallowed 404 and + // report nothing to the user. `ListStatus.executeOnPod` resolves it the same way. + if (Option(podToDelete.get()).isDefined) { getGracePeriod(sparkConf) match { case Some(period) => podToDelete.withGracePeriod(period).delete() case _ => podToDelete.delete() diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOpSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOpSuite.scala index 95a76b98b227b..30ba9040ceb00 100644 --- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOpSuite.scala +++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/submit/K8sSubmitOpSuite.scala @@ -25,7 +25,7 @@ import io.fabric8.kubernetes.api.model._ import io.fabric8.kubernetes.client.{KubernetesClient, PropagationPolicyConfigurable} import io.fabric8.kubernetes.client.dsl.{Deletable, NamespaceListVisitFromServerGetDeleteRecreateWaitApplicable, PodResource} import org.mockito.{ArgumentMatchers, Mock, MockitoAnnotations} -import org.mockito.Mockito.{times, verify, when} +import org.mockito.Mockito.{never, times, verify, when} import org.scalatest.BeforeAndAfter import org.apache.spark.{SparkConf, SparkFunSuite} @@ -38,6 +38,7 @@ import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.TEST_SP class K8sSubmitOpSuite extends SparkFunSuite with BeforeAndAfter { private val driverPodName1 = "driver1" private val driverPodName2 = "driver2" + private val missingPodName = "driver3" private val driverPod1 = buildDriverPod(driverPodName1, "1") private val driverPod2 = buildDriverPod(driverPodName2, "2") private val podList = List(driverPod1, driverPod2) @@ -55,6 +56,13 @@ class K8sSubmitOpSuite extends SparkFunSuite with BeforeAndAfter { @Mock private var driverPodOperations2: PodResource = _ + // A handle for a pod that does not exist. `withName` returns a handle regardless, and `get` + // is what reports absence by returning null, so the missing pod has to be a real mock rather + // than an unstubbed name: an unstubbed `withName` would return null and make the buggy + // `Option(handle).isDefined` check take the false branch for the wrong reason. + @Mock + private var missingPodOperations: PodResource = _ + @Mock private var kubernetesClient: KubernetesClient = _ @@ -76,10 +84,12 @@ class K8sSubmitOpSuite extends SparkFunSuite with BeforeAndAfter { when(podOperations.inNamespace(namespace)).thenReturn(podsWithNamespace) when(podsWithNamespace.withName(driverPodName1)).thenReturn(driverPodOperations1) when(podsWithNamespace.withName(driverPodName2)).thenReturn(driverPodOperations2) + when(podsWithNamespace.withName(missingPodName)).thenReturn(missingPodOperations) when(driverPodOperations1.get).thenReturn(driverPod1) when(driverPodOperations1.delete()).thenReturn(Arrays.asList(new StatusDetails)) when(driverPodOperations2.get).thenReturn(driverPod2) when(driverPodOperations2.delete()).thenReturn(Arrays.asList(new StatusDetails)) + doReturn(null).when(missingPodOperations).get } test("List app status") { @@ -120,6 +130,17 @@ class K8sSubmitOpSuite extends SparkFunSuite with BeforeAndAfter { verify(deletable, times(1)).delete() } + test("SPARK-58725: Kill app that does not exist") { + implicit val kubeClient: KubernetesClient = kubernetesClient + val killApp = new KillApplication + killApp.printStream = err + killApp.executeOnPod(missingPodName, Option(namespace), new SparkConf()) + // scalastyle:off + verify(err).println(ArgumentMatchers.eq("Application not found.")) + // scalastyle:on + verify(missingPodOperations, never()).delete() + } + test("Kill multiple apps with glob without gracePeriod") { implicit val kubeClient: KubernetesClient = kubernetesClient val killApp = new KillApplication