Skip to content

[SPARK-23948] Trigger mapstage's job listener in submitMissingTasks #21019

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
jinxing64 wants to merge 3 commits into from
Closed

[SPARK-23948] Trigger mapstage's job listener in submitMissingTasks #21019

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
685124a
[SPARK-23948] Trigger mapstage's job listener in submitMissingTasks
Apr 10, 2018
9d369f8
add a test
Apr 12, 2018
42a9b2e
Resolve comments from squito and jiangxb1987.
Apr 17, 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
Resolve comments from squito and jiangxb1987.
  • Loading branch information
jinxing committed Apr 17, 2018
commit 42a9b2e2a4f23401affbf53f207a7c98d4da0bce
5 changes: 3 additions & 2 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 @@ -2182,19 +2182,20 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
Success,
makeMapStatus("hostD", rdd2.partitions.length)))
// stage1 listener still should not have a result, though there's no missing partitions
// in it. Because stage1 is not inside runningStages at this moment.
// in it. Because stage1 has been failed and is not inside `runningStages` at this moment.
assert(listener2.results.size === 0)

// Stage0 should now be running as task set 2; make its task succeed
assert(taskSets(2).stageId === 0)
complete(taskSets(2), Seq(
(Success, makeMapStatus("hostC", rdd2.partitions.length))))
assert(mapOutputTracker.getMapSizesByExecutorId(dep1.shuffleId, 0).map(_._1).toSet ===
HashSet(makeBlockManagerId("hostC"), makeBlockManagerId("hostB")))
Set(makeBlockManagerId("hostC"), makeBlockManagerId("hostB")))

// After stage0 is finished, stage1 will be submitted and found there is no missing
// partitions in it. Then listener got triggered.
assert(listener2.results.size === 1)
assertDataStructuresEmpty()
}
Copy link
Contributor

@squito squito Apr 16, 2018

Choose a reason for hiding this comment

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

can you also add assertDataStructuresEmpty() please? I know its not really related to your change but nice to include this in all the tests.


/**
Expand Down