{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/8927","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/8927/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/8927/comments","events_url":"https://api.github.com/repos/apache/spark/issues/8927/events","html_url":"https://github.com/apache/spark/pull/8927","id":108582180,"number":8927,"title":"[SPARK-10796][CORE]Resubmit stage while lost task in Zombie and removed TaskSetsAttempt","user":{"login":"suyanNone","id":9818852,"avatar_url":"https://avatars.githubusercontent.com/u/9818852?v=3","gravatar_id":"","url":"https://api.github.com/users/suyanNone","html_url":"https://github.com/suyanNone","followers_url":"https://api.github.com/users/suyanNone/followers","following_url":"https://api.github.com/users/suyanNone/following{/other_user}","gists_url":"https://api.github.com/users/suyanNone/gists{/gist_id}","starred_url":"https://api.github.com/users/suyanNone/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/suyanNone/subscriptions","organizations_url":"https://api.github.com/users/suyanNone/orgs","repos_url":"https://api.github.com/users/suyanNone/repos","events_url":"https://api.github.com/users/suyanNone/events{/privacy}","received_events_url":"https://api.github.com/users/suyanNone/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":15,"created_at":"2015-09-28T03:23:29Z","updated_at":"2016-05-10T15:39:30Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/8927","html_url":"https://github.com/apache/spark/pull/8927","diff_url":"https://github.com/apache/spark/pull/8927.diff","patch_url":"https://github.com/apache/spark/pull/8927.patch"},"body":"We meet that problem in Spark 1.3.0, and I also reproduce on the latest version.\r\n\r\ndesc:\r\n1. We know a running `ShuffleMapStage` will have multiple `TaskSet`: one Active TaskSet, multiple Zombie TaskSet, and mutiple removedTaskSet\r\n\r\n2. We think a running `ShuffleMapStage` is success only if its partition are all process success, namely each task‘s MapStatus are all add into `outputLocs`\r\n\r\n3. MapStatus of running `ShuffleMapStage` may succeed by RemovedTaskSet1../Zombie TaskSet1 / Zombie TaskSet2 /..../ Active TaskSetN. So it had a chance that some output only hold by some RemovedTaskset or ZombieTaskSet.\r\n\r\n4. If lost a executor, it chanced that some lost-executor related MapStatus are succeed by some Zombie TaskSet.\r\n\r\n In current logical, The solution to resolved that lost MapStatus problem is,\r\n each TaskSet re-running that those tasks which succeed in lost-executor: re-add into `TaskSet's pendingTasks`, \r\nand re-add it paritions into `Stage‘s pendingPartitions` .\r\n but it is useless if that lost MapStatus only belong to Zombie/Removed TaskSet, it is Zombie, so will never be scheduled his `pendingTasks`\r\n\r\n5. The condition for resubmit stage is only if some task throws `FetchFailedException`, but may the lost-executor just not empty any MapStatus of parent Stage for one of running Stages, \r\nand it‘s happen to that running `Stage` was lost a MapStatus only belong to a ZombieTask or removedTaskset.\r\n So if all Zombie TaskSets are all processed his runningTasks and Active TaskSet are all processed his pendingTask, then will removed by `TaskSchedulerImp`, then that running Stage's pending partitions is still nonEmpty. it will hangs......\r\n\r\n\r\nTestCase to show problem:\r\n\r\n```\r\n test(\"Resubmit stage while lost partition in ZombieTasksets or RemovedTaskSets\") {\r\n    val firstRDD = new MyRDD(sc, 3, Nil)\r\n    val firstShuffleDep = new ShuffleDependency(firstRDD, new HashPartitioner(3))\r\n    val firstShuffleId = firstShuffleDep.shuffleId\r\n    val shuffleMapRdd = new MyRDD(sc, 3, List(firstShuffleDep))\r\n    val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(3))\r\n    val reduceRdd = new MyRDD(sc, 1, List(shuffleDep))\r\n    submit(reduceRdd, Array(0))\r\n\r\n    // things start out smoothly, stage 0 completes with no issues\r\n    complete(taskSets(0), Seq(\r\n      (Success, makeMapStatus(\"hostB\", shuffleMapRdd.partitions.length)),\r\n      (Success, makeMapStatus(\"hostB\", shuffleMapRdd.partitions.length)),\r\n      (Success, makeMapStatus(\"hostA\", shuffleMapRdd.partitions.length))\r\n    ))\r\n\r\n    // then start running stage 1\r\n    runEvent(makeCompletionEvent(\r\n      taskSets(1).tasks(0),\r\n      Success,\r\n      makeMapStatus(\"hostD\", shuffleMapRdd.partitions.length)))\r\n\r\n    // simulate make stage 1 resubmit, notice for stage1.0\r\n    // partitionId=1 already finished in hostD, so if we resubmit stage1,\r\n    // stage 1.1 only resubmit tasks for partitionId = 0,2\r\n    runEvent(makeCompletionEvent(\r\n      taskSets(1).tasks(1),\r\n      FetchFailed(null, firstShuffleId, 2, 1, \"Fetch failed\"), null))\r\n    scheduler.resubmitFailedStages()\r\n\r\n    val stage1Resubmit1 = taskSets(2)\r\n    assert(stage1Resubmit1.stageId == 1)\r\n    assert(stage1Resubmit1.tasks.size == 2)\r\n\r\n    // now exec-hostD lost, so the output loc of stage1 partitionId=1 will lost.\r\n    runEvent(ExecutorLost(\"exec-hostD\"))\r\n    runEvent(makeCompletionEvent(taskSets(1).tasks(0), Resubmitted, null))\r\n\r\n    // let stage1Resubmit1 complete\r\n    complete(taskSets(2), Seq(\r\n      (Success, makeMapStatus(\"hostB\", shuffleMapRdd.partitions.length)),\r\n      (Success, makeMapStatus(\"hostB\", shuffleMapRdd.partitions.length))\r\n    ))\r\n\r\n    // and let we complete tasksets1.0's active running Tasks\r\n    runEvent(makeCompletionEvent(\r\n      taskSets(1).tasks(1),\r\n      Success,\r\n      makeMapStatus(\"hostD\", shuffleMapRdd.partitions.length)))\r\n\r\n    runEvent(makeCompletionEvent(\r\n      taskSets(1).tasks(2),\r\n      Success,\r\n      makeMapStatus(\"hostD\", shuffleMapRdd.partitions.length)))\r\n\r\n    // Now all runningTasksets for stage1 was all completed. \r\n    assert(scheduler.runningStages.head.pendingPartitions.head == 0)\r\n```\r\n\r\nmain changes:\r\n1. make `DAGScheuler` only receive Task Resubmit events from ActiveTaskSets, so it can compare `pendingPartitions` with `ShuffleMapStage` missing outputs to know whether there have some partition cannot compute according current Tasksets, and make a decision if there is a need to resubmit `ShuffleMapStage`\r\n\r\nother changes:\r\n1. not register running stage outputlocs while executor lost\r\n2. Make stage's tasksets as zombie while marked as finished\r\n3. ignore expired task's partition output loc\r\n4. add make taskSetManager not handle failedTask again, which task already mark as failed due to executor lost"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/218197559","html_url":"https://github.com/apache/spark/pull/8927#issuecomment-218197559","issue_url":"https://api.github.com/repos/apache/spark/issues/8927","id":218197559,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars.githubusercontent.com/u/8108735?v=3","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2016-05-10T15:39:30Z","updated_at":"2016-05-10T15:39:30Z","body":"**[Test build #58236 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/58236/consoleFull)** for PR 8927 at commit [`3bf1eaa`](https://github.com/apache/spark/commit/3bf1eaaddc3661241b7558abd4f74cc3173aba34).\n * This patch passes all tests.\n * This patch **does not merge cleanly**.\n * This patch adds no public classes."}}