Clear all executors from the snapshot from newlyCreatedExecutors

apache · Jun 8, 2018 · bd7b0d3 · bd7b0d3
1 parent a97fc5d
commit bd7b0d3
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 7 deletions.
diff --git a/...es/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala b/...es/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
@@ -67,12 +67,7 @@ private[spark] class ExecutorPodsAllocator(
   def setTotalExpectedExecutors(total: Int): Unit = totalExpectedExecutors.set(total)
 
   private def processSnapshot(applicationId: String, snapshot: ExecutorPodsSnapshot): Unit = {
-    snapshot.executorPods.filter {
-      case (_, PodPending(_)) | (_, PodUnknown(_)) => false
-      case _ => true
-    }.keys.foreach {
-      newlyCreatedExecutors -= _
-    }
+    snapshot.executorPods.keys.foreach { newlyCreatedExecutors -= _ }
 
     // For all executors we've created against the API but have not seen in a snapshot
     // yet - check the current time. If the current time has exceeded some threshold,

diff --git a/...re/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala b/...re/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
@@ -131,7 +131,7 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
     verify(podOperations).create(podWithAttachedContainerForId(podAllocationSize + 1))
   }
 
-  test("When an executor is requested but the API does not report it in 1 minute, retry" +
+  test("When an executor is requested but the API does not report it in a reasonable time, retry" +
     " requesting that executor.") {
     podsAllocatorUnderTest.setTotalExpectedExecutors(1)
     snapshotsStore.replaceSnapshot(Seq.empty[Pod])