Skip to content

[SPARK-10515] When killing executor, the pending replacement executors should not be lost #8945

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

[SPARK-10515] When killing executor, the pending replacement executors should not be lost #8945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e559482
add unit test
Sep 30, 2015
1bdde8e
add numPendingExecutors
Sep 30, 2015
e382315
indentation style, thanks
Oct 8, 2015
b7b42cc
change test
Oct 9, 2015
cb69dc5
change test
Oct 9, 2015
da13040
Make sure the total number not changed.
Oct 13, 2015
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
2 changes: 2 additions & 0 deletions core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -438,6 +438,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
if (!replace) {
doRequestTotalExecutors(
numExistingExecutors + numPendingExecutors - executorsPendingToRemove.size)
} else {
numPendingExecutors += knownExecutors.size
}

doKillExecutors(executorsToKill)
Expand Down
35 changes: 35 additions & 0 deletions core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,6 +369,41 @@ class StandaloneDynamicAllocationSuite
assert(apps.head.getExecutorLimit === 1)
}

test("the pending replacement executors should not be lost (SPARK-10515)") {
sc = new SparkContext(appConf)
val appId = sc.applicationId
eventually(timeout(10.seconds), interval(10.millis)) {
val apps = getApplications()
assert(apps.size === 1)
assert(apps.head.id === appId)
assert(apps.head.executors.size === 2)
assert(apps.head.getExecutorLimit === Int.MaxValue)
}
// sync executors between the Master and the driver, needed because
// the driver refuses to kill executors it does not know about
syncExecutors(sc)
val executors = getExecutorIds(sc)
assert(executors.size === 2)
// kill executor 1, and replace it
assert(sc.killAndReplaceExecutor(executors.head))
eventually(timeout(10.seconds), interval(10.millis)) {
val apps = getApplications()
assert(apps.head.executors.size === 2)
}

var apps = getApplications()
// kill executor 1
assert(sc.killExecutor(executors.head))
apps = getApplications()
assert(apps.head.executors.size === 2)
Copy link
Contributor

@andrewor14 andrewor14 Oct 9, 2015

Choose a reason for hiding this comment

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

hm, not sure if I understand this. Why does the number of executors stay at 2 after you call sc.killExecutor, which does not replace it? Shouldn't it go down to 1?

Copy link
Author

@KaiXinXiaoLei KaiXinXiaoLei Oct 10, 2015

Choose a reason for hiding this comment

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

@andrewor14 Here i use sc.killExecutor(executors.head), I want to say a executor lost and a new executor should start to replace. Before the new executor registers, the executor is idle timeout. Then the total number of executors should not change. So "apps.head.executors.size === 2"

assert(apps.head.getExecutorLimit === 2)
// kill executor 2
assert(sc.killExecutor(executors(1)))
apps = getApplications()
assert(apps.head.executors.size === 1)
assert(apps.head.getExecutorLimit === 1)
}

// ===============================
// | Utility methods for testing |
// ===============================
Expand Down