[SPARK-24352][core][tests] De-flake StandaloneDynamicAllocationSuite blacklist test

core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala

@@ -493,28 +493,35 @@ class StandaloneDynamicAllocationSuite
   }
 
   test("executor registration on a blacklisted host must fail") {
+    // The context isn't really used by the test, but it helps with creating a test scheduler,
+    // since CoarseGrainedSchedulerBackend makes a lot of calls to the context instance.
     sc = new SparkContext(appConf.set(config.BLACKLIST_ENABLED.key, "true"))
+
     val endpointRef = mock(classOf[RpcEndpointRef])
     val mockAddress = mock(classOf[RpcAddress])
     when(endpointRef.address).thenReturn(mockAddress)
-    val message = RegisterExecutor("one", endpointRef, "blacklisted-host", 10, Map.empty, Map.empty,
-      Map.empty)
+    val message = RegisterExecutor("one", endpointRef, "blacklisted-host", 10, Map.empty,
+      Map.empty, Map.empty)
 
-    // Get "localhost" on a blacklist.
     val taskScheduler = mock(classOf[TaskSchedulerImpl])
     when(taskScheduler.nodeBlacklist()).thenReturn(Set("blacklisted-host"))
+    when(taskScheduler.resourceOffers(any())).thenReturn(Nil)
     when(taskScheduler.sc).thenReturn(sc)
-    sc.taskScheduler = taskScheduler
-
-    // Create a fresh scheduler backend to blacklist "localhost".
-    sc.schedulerBackend.stop()
-    val backend =
-      new StandaloneSchedulerBackend(taskScheduler, sc, Array(masterRpcEnv.address.toSparkURL))
-    backend.start()
 
-    backend.driverEndpoint.ask[Boolean](message)
-    eventually(timeout(10.seconds), interval(100.millis)) {
-      verify(endpointRef).send(RegisterExecutorFailed(any()))
+    val rpcEnv = RpcEnv.create("test-rpcenv", "localhost", 0, conf, securityManager)
+    try {
+      val scheduler = new CoarseGrainedSchedulerBackend(taskScheduler, rpcEnv)
+      try {
+        scheduler.start()
+        scheduler.driverEndpoint.ask[Boolean](message)
+        eventually(timeout(10.seconds), interval(100.millis)) {
+          verify(endpointRef).send(RegisterExecutorFailed(any()))
+        }
+      } finally {
+        scheduler.stop()
+      }
+    } finally {
+      rpcEnv.shutdown()
     }
   }