[SPARK-30440][CORE][TESTS] Avoid race condition in TaskSetManagerSuit…

…e by not using resourceOffer

### What changes were proposed in this pull request?
There is a race condition in test case introduced in SPARK-30359 between reviveOffers in org.apache.spark.scheduler.TaskSchedulerImpl#submitTasks and org.apache.spark.scheduler.TaskSetManager#resourceOffer, in the testcase

No need to do resourceOffers as submitTask will revive offers from task set

### Why are the changes needed?
Fix flaky test

### Does this PR introduce any user-facing change?
No

### How was this patch tested?
Test case can pass after the change

Closes apache#27115 from ajithme/testflaky.

Authored-by: Ajith <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala

@@ -1940,27 +1940,27 @@ class TaskSetManagerSuite
     val backend = sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
 
     TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
-    val Seq(exec0, exec1) = backend.getExecutorIds()
 
-    val taskSet = FakeTask.createTaskSet(2)
+    val tasks = Array.tabulate[Task[_]](2)(partition => new FakeLongTasks(stageId = 0, partition))
+    val taskSet: TaskSet = new TaskSet(tasks, stageId = 0, stageAttemptId = 0, priority = 0, null)
     val stageId = taskSet.stageId
     val stageAttemptId = taskSet.stageAttemptId
     sched.submitTasks(taskSet)
-    val taskSetManagers = PrivateMethod[mutable.HashMap[Int, mutable.HashMap[Int, TaskSetManager]]](
-      Symbol("taskSetsByStageIdAndAttempt"))
+    val taskSetManagers =
+      PrivateMethod[mutable.HashMap[Int, mutable.HashMap[Int, TaskSetManager]]](
+        Symbol("taskSetsByStageIdAndAttempt"))
     // get the TaskSetManager
     val manager = sched.invokePrivate(taskSetManagers()).get(stageId).get(stageAttemptId)
 
-    val task0 = manager.resourceOffer(exec0, "localhost", TaskLocality.NO_PREF)
-    val task1 = manager.resourceOffer(exec1, "localhost", TaskLocality.NO_PREF)
-    assert(task0.isDefined && task1.isDefined)
-    val (taskId0, index0) = (task0.get.taskId, task0.get.index)
-    val (taskId1, index1) = (task1.get.taskId, task1.get.index)
+    val (task0, task1) = eventually(timeout(10.seconds), interval(100.milliseconds)) {
+      (manager.taskInfos(0), manager.taskInfos(1))
+    }
+
+    val (taskId0, index0, exec0) = (task0.taskId, task0.index, task0.executorId)
+    val (taskId1, index1, exec1) = (task1.taskId, task1.index, task1.executorId)
     // set up two running tasks
     assert(manager.taskInfos(taskId0).running)
-    assert(manager.taskInfos(taskId0).executorId === exec0)
     assert(manager.taskInfos(taskId1).running)
-    assert(manager.taskInfos(taskId1).executorId === exec1)
 
     val numFailures = PrivateMethod[Array[Int]](Symbol("numFailures"))
     // no task failures yet
@@ -1981,3 +1981,13 @@ class TaskSetManagerSuite
     }
   }
 }
+
+class FakeLongTasks(stageId: Int, partitionId: Int) extends FakeTask(stageId, partitionId) {
+
+  override def runTask(context: TaskContext): Int = {
+    while (true) {
+      Thread.sleep(10000)
+    }
+    0
+  }
+}