simulate executor registration and test executor lifecycle properly

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

Author: sudiptob2

src/test/scala/org/apache/spark/scheduler/cluster/armada/ArmadaDynamicAllocationIntegrationSuite.scala

@@ -88,7 +88,7 @@ class ArmadaDynamicAllocationIntegrationSuite extends AnyFunSuite with BeforeAnd
     sc = createMockSparkContext(conf)
     val taskScheduler = createMockTaskScheduler(sc)
 
-    backend = new ArmadaClusterManagerBackend(
+    backend = new TestableArmadaClusterManagerBackend(
       taskScheduler,
       sc,
       java.util.concurrent.Executors.newScheduledThreadPool(1),
@@ -113,6 +113,13 @@ class ArmadaDynamicAllocationIntegrationSuite extends AnyFunSuite with BeforeAnd
       backend.getPendingExecutorCount === 1,
       "Executor should stay pending until registered with Spark"
     )
+
+    backend.asInstanceOf[TestableArmadaClusterManagerBackend].simulateExecutorRegistration(execId)
+
+    assert(
+      backend.getPendingExecutorCount === 0,
+      "Executor should be removed from pending after registering with Spark"
+    )
   }
 
   test("concurrent executor submissions are thread-safe") {
@@ -183,6 +190,26 @@ class ArmadaDynamicAllocationIntegrationSuite extends AnyFunSuite with BeforeAnd
 
   // Helper methods
 
+  private class TestableArmadaClusterManagerBackend(
+      scheduler: TaskSchedulerImpl,
+      sc: SparkContext,
+      executorService: java.util.concurrent.ScheduledExecutorService,
+      masterURL: String
+  ) extends ArmadaClusterManagerBackend(scheduler, sc, executorService, masterURL) {
+
+    private val testRegisteredExecutors = scala.collection.mutable.Set.empty[String]
+
+    /** Simulate executor registration by adding it to the test set */
+    def simulateExecutorRegistration(executorId: String): Unit = {
+      testRegisteredExecutors += executorId
+    }
+
+    /** Include our test executors */
+    override def getExecutorIds(): Seq[String] = synchronized {
+      super.getExecutorIds() ++ testRegisteredExecutors.toSeq
+    }
+  }
+
   private def createMockSparkContext(sparkConf: SparkConf): SparkContext = {
     val sc     = org.mockito.Mockito.mock(classOf[SparkContext])
     val env    = org.mockito.Mockito.mock(classOf[SparkEnv])