Delete useless code

Author: xuanyuanking

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

@@ -800,18 +800,6 @@ private[spark] class TaskSetManager(
             fetchFailed.bmAddress.host, fetchFailed.bmAddress.executorId))
         }
         fetchFailedTaskIndexSet.add(index)
-        // Kill any other attempts for this FetchFailed task
-        for (attemptInfo <- taskAttempts(index) if attemptInfo.running) {
-          logInfo(s"Killing attempt ${attemptInfo.attemptNumber} for task ${attemptInfo.id} " +
-            s"in stage ${taskSet.id} (TID ${attemptInfo.taskId}) on ${attemptInfo.host} " +
-            s"as the attempt ${info.attemptNumber} failed because FetchFailed")
-          killedByOtherAttempt(index) = true
-          sched.backend.killTask(
-            attemptInfo.taskId,
-            attemptInfo.executorId,
-            interruptThread = true,
-            reason = "another attempt fetch failed")
-        }
 
         None
 

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

@@ -852,97 +852,6 @@ class TaskSetManagerSuite extends SparkFunSuite with LocalSparkContext with Logg
     assert(resubmittedTasks === 0)
   }
 
-  test("Fetch failed task should not have success completion event") {
-    sc = new SparkContext("local", "test")
-    // Set the speculation multiplier to be 0 so speculative tasks are launched immediately
-    sc.conf.set("spark.speculation.multiplier", "0.0")
-    sc.conf.set("spark.speculation", "true")
-
-    sched = new FakeTaskScheduler(sc, ("exec1", "host1"), ("exec2", "host2"))
-    var killTaskCalled = false
-    sched.initialize(new FakeSchedulerBackend() {
-      override def killTask(taskId: Long,
-          executorId: String,
-          interruptThread: Boolean,
-          reason: String): Unit = {
-        // Check the only one killTask event in this case, which triggered by
-        // task 3.1 fetch failed.
-        assert(taskId === 3)
-        assert(executorId === "exec2")
-        assert(interruptThread)
-        assert(reason === "another attempt fetch failed")
-        killTaskCalled = true
-      }
-    })
-
-    // Keep track of the original tasks will not has SUCCESS event.
-    var originTaskSuccess = false
-    val dagScheduler = new FakeDAGScheduler(sc, sched) {
-      override def taskEnded(task: Task[_],
-          reason: TaskEndReason,
-          result: Any,
-          accumUpdates: Seq[AccumulatorV2[_, _]],
-          taskInfo: TaskInfo): Unit = {
-        super.taskEnded(task, reason, result, accumUpdates, taskInfo)
-        reason match {
-          case Success if taskInfo.taskId == 3 &&
-              taskInfo.attemptNumber == 0 && taskInfo.index == 3 =>
-            originTaskSuccess = true
-          case _ =>
-        }
-      }
-    }
-    sched.dagScheduler.stop()
-    sched.setDAGScheduler(dagScheduler)
-
-    val taskSet = FakeTask.createTaskSet(4)
-    val clock = new ManualClock()
-    val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES, clock = clock)
-    val accumUpdatesByTask: Array[Seq[AccumulatorV2[_, _]]] = taskSet.tasks.map { task =>
-      task.metrics.internalAccums
-    }
-    // Offer resources for 4 tasks to start
-    for ((k, v) <- List(
-      "exec1" -> "host1",
-      "exec1" -> "host1",
-      "exec2" -> "host2",
-      "exec2" -> "host2")) {
-      val taskOption = manager.resourceOffer(k, v, NO_PREF)
-      assert(taskOption.isDefined)
-      val task = taskOption.get
-      assert(task.executorId === k)
-    }
-    assert(sched.startedTasks.toSet === Set(0, 1, 2, 3))
-    clock.advance(1)
-    // Complete the 3 tasks and leave 1 task in running
-    for (id <- Set(0, 1, 2)) {
-      manager.handleSuccessfulTask(id, createTaskResult(id, accumUpdatesByTask(id)))
-      assert(sched.endedTasks(id) === Success)
-    }
-
-    // checkSpeculatableTasks checks that the task runtime is greater than the threshold for
-    // speculating. Since we use a threshold of 0 for speculation, tasks need to be running for
-    // > 0ms, so advance the clock by 1ms here.
-    clock.advance(1)
-    assert(manager.checkSpeculatableTasks(0))
-    assert(sched.speculativeTasks.toSet === Set(3))
-
-    // Offer resource to start the speculative attempt for the running task
-    val taskOption5 = manager.resourceOffer("exec1", "host1", NO_PREF)
-    assert(taskOption5.isDefined)
-    val task5 = taskOption5.get
-    assert(task5.index === 3)
-    assert(task5.taskId === 4)
-    assert(task5.executorId === "exec1")
-    assert(task5.attemptNumber === 1)
-    // sched.backend = mock(classOf[SchedulerBackend])
-    // The speculative task raise a FetchFailed
-    manager.handleFailedTask(4, TaskState.FAILED,
-      FetchFailed(BlockManagerId("exec1", "host1", 12345), 0, 0, 0, "ignored"))
-    // Check the original task killed by FetchFailed.
-    assert(!originTaskSuccess && killTaskCalled)
-  }
-
   test("speculative and noPref task should be scheduled after node-local") {
     sc = new SparkContext("local", "test")
     sched = new FakeTaskScheduler(