Skip to content

[SPARK-17644] [CORE] Do not add failedStages when abortStage for fetch failure #15213

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
scwf wants to merge 9 commits into from
Closed

[SPARK-17644] [CORE] Do not add failedStages when abortStage for fetch failure #15213

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
8e667f5
test case
scwf Sep 23, 2016
2bfa05b
The failed stage never resubmitted
scwf Sep 23, 2016
d02cf93
test case
scwf Sep 23, 2016
7056cd6
fix style
scwf Sep 23, 2016
1f7bd88
test case imporvement
scwf Sep 23, 2016
d92adfc
comment fix
scwf Sep 23, 2016
1127ca1
address comments of zsxwing
scwf Sep 23, 2016
f91d86f
improve test
scwf Sep 28, 2016
09077cb
comment fix
scwf Sep 28, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
comment fix
  • Loading branch information
@scwf
scwf committed Sep 23, 2016
commit d92adfcd2c51090dd472a995b5853cd6396aeee7
18 changes: 8 additions & 10 deletions core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2119,12 +2119,11 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with Timeou
val rdd1 = sc.makeRDD(Array(1, 2, 3, 4), 2).map(x => (x, 1)).groupByKey()
val shuffleHandle =
rdd1.dependencies.head.asInstanceOf[ShuffleDependency[_, _, _]].shuffleHandle
rdd1.map { x =>
if (x._1 == 1) {
throw new FetchFailedException(
BlockManagerId("1", "1", 1), shuffleHandle.shuffleId, 0, 0, "test")
}
x._1
rdd1.map {
case (x, _) if (x == 1) =>
throw new FetchFailedException(
BlockManagerId("1", "1", 1), shuffleHandle.shuffleId, 0, 0, "test")
case (x, _) => x
}.count()
} catch {
case e: Throwable =>
Copy link
Member

@zsxwing zsxwing Sep 23, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

don't catch Throwable and it's better to check the exception like this:

    failAfter(60.seconds) {
      val e = intercept[SparkException] {
        ...
      }
      assert(e.getMessage.contains("org.apache.spark.shuffle.FetchFailedException"))
    }

Expand All @@ -2137,12 +2136,11 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with Timeou
val rdd2 = sc.makeRDD(Array(1, 2, 3, 4), 2).map(x => (x, 1)).groupByKey()
val shuffleHandle =
rdd2.dependencies.head.asInstanceOf[ShuffleDependency[_, _, _]].shuffleHandle
rdd2.map { x =>
if (x._1 == 1) {
rdd2.map {
case (x, _) if (x == 1) =>
throw new FetchFailedException(
BlockManagerId("1", "1", 1), shuffleHandle.shuffleId, 0, 0, "test")
}
x._1
case (x, _) => x
}.count()
} catch {
case e: Throwable =>
Expand Down