Skip to content

[SPARK-13343] speculative tasks that didn't commit shouldn't be marked as success #21653

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
hthuynh2 wants to merge 8 commits into from
Closed

[SPARK-13343] speculative tasks that didn't commit shouldn't be marked as success #21653

Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8f7d981
Fixed issue and added unit test
Jun 28, 2018
980a933
Fixed Scala Style Error
hthuynh2 Jun 28, 2018
7db26f2
Modified Unit Test and Kill Message
Jul 6, 2018
f66fcab
Merge branch 'SPARK_13343' of https://github.com/hthuynh2/spark into …
Jul 6, 2018
3c655f2
Fix scala Style
Jul 6, 2018
7d9e4bb
Merge branch 'master' of https://github.com/apache/spark into SPARK_1…
Jul 19, 2018
2c7d33d
undo effect on totalResultSize and calculatedTasks
Jul 19, 2018
b6585da
add more comments
Jul 23, 2018
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
undo effect on totalResultSize and calculatedTasks
  • Loading branch information
Hieu Huynh authored and Hieu Huynh committed Jul 19, 2018
commit 2c7d33d88a9023604ab6343988f3263ceea333ca
12 changes: 10 additions & 2 deletions core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import org.apache.spark._
import org.apache.spark.TaskState.TaskState
import org.apache.spark.internal.{config, Logging}
import org.apache.spark.scheduler.SchedulingMode._
import org.apache.spark.util.{AccumulatorV2, Clock, SystemClock, Utils}
import org.apache.spark.util.{AccumulatorV2, Clock, LongAccumulator, SystemClock, Utils}
import org.apache.spark.util.collection.MedianHeap

/**
Expand Down Expand Up @@ -724,7 +724,15 @@ private[spark] class TaskSetManager(
val info = taskInfos(tid)
val index = info.index
// Check if any other attempt succeeded before this and this attempt has not been handled
if (successful(index) && killedByOtherAttempt(index)) {
if (successful(index) && killedByOtherAttempt.contains(tid)) {
calculatedTasks -= 1
Copy link
Contributor

@tgravescs tgravescs Jul 20, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

please add a comment here about cleaning up things from incremented earlier while handling it as successful


val resultSizeAcc = result.accumUpdates.find(a =>
a.name == Some(InternalAccumulator.RESULT_SIZE))
if (resultSizeAcc.isDefined) {
totalResultSize -= resultSizeAcc.get.asInstanceOf[LongAccumulator].value
Copy link
Contributor

@tgravescs tgravescs Jul 20, 2018

Choose a reason for hiding this comment

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

the downside here is we already incremented and other tasks could have checked and failed before we decrement, but unless someone else has a better idea this is better then it is now.

Copy link
Contributor

@mridulm mridulm Jul 21, 2018

Choose a reason for hiding this comment

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

I agree, I dont see a better option.

}

handleFailedTask(tid, TaskState.KILLED,
TaskKilled("Finish but did not commit due to another attempt succeeded"))
return
Expand Down