{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/35604775","id":35604775,"diff_hunk":"@@ -473,6 +473,232 @@ class DAGSchedulerSuite\n     assertDataStructuresEmpty()\n   }\n \n+  // Helper function to validate state and print output when creating tests for task failures\n+  def checkStageIdAndPrint(stageId: Int, attempt: Int, stageAttempt: TaskSet) {\n+    println(s\"$attempt($attempt): taskSets = $taskSets : ${\n+      taskSets.map{_.tasks.mkString(\",\")}.mkString(\";\")}\")\n+\n+    assert(stageAttempt.stageId === stageId)\n+    assert(stageAttempt.stageAttemptId == attempt)\n+    println(s\"tasks for $stageAttempt : ${stageAttempt.tasks.mkString(\",\")}\")\n+  }\n+\n+  /**\n+   * In this test we simulate a job failure where the first stage completes successfully and\n+   * the second stage fails due to a fetch failure. Multiple successive fetch failures of a stage\n+   * trigger an overall stage abort to avoid endless retries.\n+   */\n+  test(\"Multiple consecutive stage failures should lead to task being aborted.\") {\n+    // Create a new Listener to confirm that the listenerBus sees the JobEnd message\n+    // when we abort the stage. This message will also be consumed by the EventLoggingListener\n+    // so this will propagate up to the user.\n+    var ended = false\n+    var jobResult : JobResult = null\n+    class EndListener extends SparkListener {\n+      override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {\n+        jobResult = jobEnd.jobResult\n+        ended = true\n+      }\n+    }\n+\n+    sc.listenerBus.addListener(new EndListener())\n+\n+    val shuffleMapRdd = new MyRDD(sc, 2, Nil)\n+    val shuffleDep = new ShuffleDependency(shuffleMapRdd, null)\n+    val shuffleId = shuffleDep.shuffleId\n+    val reduceRdd = new MyRDD(sc, 2, List(shuffleDep))\n+    submit(reduceRdd, Array(0, 1))\n+\n+    for (attempt <- 0 to Stage.MAX_STAGE_FAILURES) {\n+      // Complete all the tasks for the current attempt of stage 0 successfully\n+      val stage0Attempt = taskSets.last\n+\n+      // Confirm  that this is the first attempt for stage 0\n+      checkStageIdAndPrint(0, attempt, stage0Attempt)\n+\n+      // Make each task in stage 0 success\n+      val completions = stage0Attempt.tasks.zipWithIndex.map{ case (task, idx) =>\n+        (Success, makeMapStatus(\"host\" + ('A' + idx).toChar, 2))\n+      }.toSeq\n+\n+      // Run stage 0\n+      complete(stage0Attempt, completions)\n+\n+      // Now we should have a new taskSet, for a new attempt of stage 1.\n+      // We will have one fetch failure for this task set\n+      val stage1Attempt = taskSets.last\n+      checkStageIdAndPrint(1, attempt, stage1Attempt)\n+\n+      val stage1Successes = stage1Attempt.tasks.tail.map { _ => (Success, 42)}\n+\n+      // Run Stage 1, this time with a task failure\n+      complete(stage1Attempt,\n+        Seq((FetchFailed(makeBlockManagerId(\"hostA\"), shuffleId, 0, 0, \"ignored\"), null))\n+          ++ stage1Successes\n+      )\n+\n+      // this will (potentially) trigger a resubmission of stage 0, since we've lost some of its\n+      // map output, for the next iteration through the loop\n+      scheduler.resubmitFailedStages()\n+\n+      if (attempt < Stage.MAX_STAGE_FAILURES) {\n+        assert(scheduler.runningStages.nonEmpty)\n+        assert(!ended)\n+      } else {\n+        // Stage has been aborted and removed from running stages\n+        assertDataStructuresEmpty()\n+        sc.listenerBus.waitUntilEmpty(1000)\n+        assert(ended)\n+        assert(jobResult.isInstanceOf[JobFailed])\n+      }\n+    }\n+  }\n+\n+  /**\n+   * In this test we simulate a job failure where there are two failures in two different stages.\n+   * Specifically, stage0 fails twice, and then stage1 twice. In total, the job has had four\n+   * failures overall but not four failures for a particular stage, and as such should not be\n+   * aborted.\n+   */\n+  test(\"Failures in different stages should not trigger an overall abort\") {\n+    // Create a new Listener to confirm that the listenerBus sees the JobEnd message\n+    // when we abort the stage. This message will also be consumed by the EventLoggingListener\n+    // so this will propagate up to the user.\n+    var ended = false\n+    var jobResult : JobResult = null\n+    class EndListener extends SparkListener {\n+      override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {\n+        jobResult = jobEnd.jobResult\n+        ended = true\n+      }\n+    }\n+\n+    sc.listenerBus.addListener(new EndListener())\n+\n+    val shuffleMapRdd = new MyRDD(sc, 2, Nil)\n+    val shuffleDep = new ShuffleDependency(shuffleMapRdd, null)\n+    val shuffleId = shuffleDep.shuffleId\n+    val reduceRdd = new MyRDD(sc, 2, List(shuffleDep))\n+    submit(reduceRdd, Array(0, 1))\n+\n+    for (attempt <- 0 to Stage.MAX_STAGE_FAILURES) {\n+      // Complete all the tasks for the current attempt of stage 0 successfully\n+      val stage0Attempt = taskSets.last\n+\n+      // Confirm  that this is the first attempt for stage 0\n+      checkStageIdAndPrint(0, attempt, stage0Attempt)\n+\n+      if (attempt < Stage.MAX_STAGE_FAILURES/2) {\n+        // Make each task in stage 0 success\n+        val completions = stage0Attempt.tasks.zipWithIndex.map{ case (task, idx) =>\n+          (Success, makeMapStatus(\"host\" + ('A' + idx).toChar, 2))\n+        }.toSeq\n+\n+        // Run stage 0\n+        complete(stage0Attempt, completions)\n+\n+        // Now we should have a new taskSet, for a new attempt of stage 1.\n+        // We will have one fetch failure for this task set\n+        val stage1Attempt = taskSets.last\n+        checkStageIdAndPrint(1, attempt, stage1Attempt)\n+\n+        val stage1Successes = stage1Attempt.tasks.tail.map { _ => (Success, 42)}\n+\n+        // Run Stage 1, this time with a task failure\n+        complete(stage1Attempt,\n+          Seq((FetchFailed(makeBlockManagerId(\"hostA\"), shuffleId, 0, 0, \"ignored\"), null))\n+            ++ stage1Successes\n+        )\n+      } else {\n+        val stage0Successes = stage0Attempt.tasks.tail.map { _ => (Success, 42)}\n+\n+        // Run stage 0 and fail\n+        complete(stage0Attempt,\n+          Seq((FetchFailed(makeBlockManagerId(\"hostA\"), shuffleId, 0, 0, \"ignored\"), null))\n+            ++ stage0Successes\n+        )\n+      }\n+\n+      // this will (potentially) trigger a resubmission of stage 0, since we've lost some of its\n+      // map output, for the next iteration through the loop\n+      scheduler.resubmitFailedStages()\n+\n+      // In this test case, we're confirming that we never actually fail out (so we should have a\n+      // stage that has been resubmitted)\n+      assert(scheduler.runningStages.nonEmpty)\n+      assert(!ended)\n+    }","path":"core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala","position":158,"original_position":158,"commit_id":"34d69fae8fab3f4b778a5d986dd335f8cca239ba","original_commit_id":"34d69fae8fab3f4b778a5d986dd335f8cca239ba","user":{"login":"squito","id":71240,"avatar_url":"https://avatars.githubusercontent.com/u/71240?v=3","gravatar_id":"","url":"https://api.github.com/users/squito","html_url":"https://github.com/squito","followers_url":"https://api.github.com/users/squito/followers","following_url":"https://api.github.com/users/squito/following{/other_user}","gists_url":"https://api.github.com/users/squito/gists{/gist_id}","starred_url":"https://api.github.com/users/squito/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/squito/subscriptions","organizations_url":"https://api.github.com/users/squito/orgs","repos_url":"https://api.github.com/users/squito/repos","events_url":"https://api.github.com/users/squito/events{/privacy}","received_events_url":"https://api.github.com/users/squito/received_events","type":"User","site_admin":false},"body":"I just want to make sure that we can actually finish the job successfully -- you could imagine your checks all passing, but the job is stuck in some weird state where the DAGSCheduler thinks the stage is still running, but no task sets actually created, or something weird like that.  Your asserts are almost there, but not quite.  the final call to `scheduler.resubmitFailedStages` should trigger another attempt for stage 0, and after completing that, you'd get another attempt for stage 1, and then finally the job should complete, but successfully.\r\n\r\nI know this is minor, I'd just like to be extra thorough.","created_at":"2015-07-28T00:43:55Z","updated_at":"2015-07-28T00:43:55Z","html_url":"https://github.com/apache/spark/pull/5636#discussion_r35604775","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/5636","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/35604775"},"html":{"href":"https://github.com/apache/spark/pull/5636#discussion_r35604775"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/5636"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/5636","id":33885433,"html_url":"https://github.com/apache/spark/pull/5636","diff_url":"https://github.com/apache/spark/pull/5636.diff","patch_url":"https://github.com/apache/spark/pull/5636.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/5636","number":5636,"state":"open","locked":false,"title":"[SPARK-5945] Spark should not retry a stage infinitely on a FetchFailedException","user":{"login":"ilganeli","id":1483345,"avatar_url":"https://avatars.githubusercontent.com/u/1483345?v=3","gravatar_id":"","url":"https://api.github.com/users/ilganeli","html_url":"https://github.com/ilganeli","followers_url":"https://api.github.com/users/ilganeli/followers","following_url":"https://api.github.com/users/ilganeli/following{/other_user}","gists_url":"https://api.github.com/users/ilganeli/gists{/gist_id}","starred_url":"https://api.github.com/users/ilganeli/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilganeli/subscriptions","organizations_url":"https://api.github.com/users/ilganeli/orgs","repos_url":"https://api.github.com/users/ilganeli/repos","events_url":"https://api.github.com/users/ilganeli/events{/privacy}","received_events_url":"https://api.github.com/users/ilganeli/received_events","type":"User","site_admin":false},"body":"All - I've added a map to track the reasons, and counts / reason for stage failures. I then check whether to abort a stage based on whether there is a sufficient number of failures for a single reason.\r\n\r\nThe open questions are:\r\n\r\n1)  Is it really safe to assume that a ```FetchFailedException``` means that the ```BlockManager``` has failed, and it will work if we just try another one? E.g. does retrying the failed stage even make sense?\r\n2) For the equality check, I'm using the ```failureMessage``` inside the handler for ```FetchFailed```. Is it safe to assume that string will be consistent between subsequent stage failures or are there any counters or such in there that will change? If this isn't a safe item to use for comparison, is there any metric that can be used to track the number of failures for a particular reason?\r\n3) How many attempts for a stage should be made before aborting? I know that a task can have up to 4  failures before it's aborted but since a stage may see multiple fetch failures from concurrent tasks, I assume this number should be higher. ","created_at":"2015-04-22T18:14:08Z","updated_at":"2015-07-28T00:43:55Z","closed_at":null,"merged_at":null,"merge_commit_sha":"b12d7baaa44cc9dc69f090361d74b278dfd5c5e3","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/5636/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/5636/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/5636/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/34d69fae8fab3f4b778a5d986dd335f8cca239ba","head":{"label":"ilganeli:SPARK-5945","ref":"SPARK-5945","sha":"34d69fae8fab3f4b778a5d986dd335f8cca239ba","user":{"login":"ilganeli","id":1483345,"avatar_url":"https://avatars.githubusercontent.com/u/1483345?v=3","gravatar_id":"","url":"https://api.github.com/users/ilganeli","html_url":"https://github.com/ilganeli","followers_url":"https://api.github.com/users/ilganeli/followers","following_url":"https://api.github.com/users/ilganeli/following{/other_user}","gists_url":"https://api.github.com/users/ilganeli/gists{/gist_id}","starred_url":"https://api.github.com/users/ilganeli/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilganeli/subscriptions","organizations_url":"https://api.github.com/users/ilganeli/orgs","repos_url":"https://api.github.com/users/ilganeli/repos","events_url":"https://api.github.com/users/ilganeli/events{/privacy}","received_events_url":"https://api.github.com/users/ilganeli/received_events","type":"User","site_admin":false},"repo":{"id":24374828,"name":"spark","full_name":"ilganeli/spark","owner":{"login":"ilganeli","id":1483345,"avatar_url":"https://avatars.githubusercontent.com/u/1483345?v=3","gravatar_id":"","url":"https://api.github.com/users/ilganeli","html_url":"https://github.com/ilganeli","followers_url":"https://api.github.com/users/ilganeli/followers","following_url":"https://api.github.com/users/ilganeli/following{/other_user}","gists_url":"https://api.github.com/users/ilganeli/gists{/gist_id}","starred_url":"https://api.github.com/users/ilganeli/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ilganeli/subscriptions","organizations_url":"https://api.github.com/users/ilganeli/orgs","repos_url":"https://api.github.com/users/ilganeli/repos","events_url":"https://api.github.com/users/ilganeli/events{/privacy}","received_events_url":"https://api.github.com/users/ilganeli/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/ilganeli/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/ilganeli/spark","forks_url":"https://api.github.com/repos/ilganeli/spark/forks","keys_url":"https://api.github.com/repos/ilganeli/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ilganeli/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ilganeli/spark/teams","hooks_url":"https://api.github.com/repos/ilganeli/spark/hooks","issue_events_url":"https://api.github.com/repos/ilganeli/spark/issues/events{/number}","events_url":"https://api.github.com/repos/ilganeli/spark/events","assignees_url":"https://api.github.com/repos/ilganeli/spark/assignees{/user}","branches_url":"https://api.github.com/repos/ilganeli/spark/branches{/branch}","tags_url":"https://api.github.com/repos/ilganeli/spark/tags","blobs_url":"https://api.github.com/repos/ilganeli/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ilganeli/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ilganeli/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/ilganeli/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ilganeli/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/ilganeli/spark/languages","stargazers_url":"https://api.github.com/repos/ilganeli/spark/stargazers","contributors_url":"https://api.github.com/repos/ilganeli/spark/contributors","subscribers_url":"https://api.github.com/repos/ilganeli/spark/subscribers","subscription_url":"https://api.github.com/repos/ilganeli/spark/subscription","commits_url":"https://api.github.com/repos/ilganeli/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/ilganeli/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/ilganeli/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/ilganeli/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/ilganeli/spark/contents/{+path}","compare_url":"https://api.github.com/repos/ilganeli/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ilganeli/spark/merges","archive_url":"https://api.github.com/repos/ilganeli/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ilganeli/spark/downloads","issues_url":"https://api.github.com/repos/ilganeli/spark/issues{/number}","pulls_url":"https://api.github.com/repos/ilganeli/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/ilganeli/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/ilganeli/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ilganeli/spark/labels{/name}","releases_url":"https://api.github.com/repos/ilganeli/spark/releases{/id}","created_at":"2014-09-23T14:30:54Z","updated_at":"2015-04-08T21:51:24Z","pushed_at":"2015-07-27T20:49:25Z","git_url":"git://github.com/ilganeli/spark.git","ssh_url":"git@github.com:ilganeli/spark.git","clone_url":"https://github.com/ilganeli/spark.git","svn_url":"https://github.com/ilganeli/spark","homepage":null,"size":111796,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"3ab7525dceeb1c2f3c21efb1ee5a9c8bb0fd0c13","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","created_at":"2014-02-25T08:00:08Z","updated_at":"2015-07-27T21:50:45Z","pushed_at":"2015-07-28T00:32:48Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":1601512,"stargazers_count":4873,"watchers_count":4873,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":4402,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":346,"forks":4402,"open_issues":346,"watchers":4873,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/5636"},"html":{"href":"https://github.com/apache/spark/pull/5636"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/5636"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/5636/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/5636/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/5636/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/34d69fae8fab3f4b778a5d986dd335f8cca239ba"}}}}