Skip to content

[SPARK-28455][core] Avoid overflow when calculating executor timeout time #25208

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

[SPARK-28455][core] Avoid overflow when calculating executor timeout time #25208

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
adeae39
[SPARK-28455][core] Avoid overflow when calculating executor timeout …
Jul 19, 2019
392e585
Log tweak.
Jul 22, 2019
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
19 changes: 15 additions & 4 deletions core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,12 @@ private[spark] class ExecutorMonitor(
exec.updateActiveShuffles(activeShuffleIds)
if (exec.hasActiveShuffle) {
needTimeoutUpdate = true
activatedExecs += id
// Don't collect too many ids or the debug message becomes not very useful.
if (activatedExecs.size < 10) {

@mgaido91 mgaido91 Jul 20, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

what about having a method for this in order to avoid code duplication?

activatedExecs += id
} else if (activatedExecs.size == 10) {
activatedExecs += "..."
}

@dongjoon-hyun dongjoon-hyun Jul 20, 2019 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Shall we wrap this with if (log.isDebugEnabled) { ... }, too?

}
}
}
Expand Down Expand Up @@ -237,8 +242,13 @@ private[spark] class ExecutorMonitor(
executors.asScala.foreach { case (id, exec) =>
if (exec.hasActiveShuffle) {
exec.updateActiveShuffles(activeShuffles)
if (!exec.hasActiveShuffle) {
deactivatedExecs += id
if (!exec.hasActiveShuffle && log.isDebugEnabled) {
// Don't collect too many ids or the debug message becomes not very useful.
if (deactivatedExecs.size < 10) {
deactivatedExecs += id
} else if (deactivatedExecs.size == 10) {
deactivatedExecs += "..."
}
}
}
}
Expand Down Expand Up @@ -448,7 +458,8 @@ private[spark] class ExecutorMonitor(
} else {
idleTimeoutMs
}
idleStart + timeout
val deadline = idleStart + timeout
if (deadline >= 0) deadline else Long.MaxValue
} else {
Long.MaxValue
}
Expand Down
20 changes: 20 additions & 0 deletions core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -367,6 +367,26 @@ class ExecutorMonitorSuite extends SparkFunSuite {
assert(monitor.timedOutExecutors(idleDeadline).toSet === Set("1", "2"))
}

test("SPARK-28455: avoid overflow in timeout calculation") {
conf
.set(DYN_ALLOCATION_SHUFFLE_TIMEOUT, Long.MaxValue)
.set(DYN_ALLOCATION_SHUFFLE_TRACKING, true)
.set(SHUFFLE_SERVICE_ENABLED, false)
monitor = new ExecutorMonitor(conf, client, null, clock)

// Generate events that will make executor 1 be idle, while still holding shuffle data.
// The executor should not be eligible for removal since the timeout is basically "infinite".
val stage = stageInfo(1, shuffleId = 0)
monitor.onJobStart(SparkListenerJobStart(1, clock.getTimeMillis(), Seq(stage)))
clock.advance(1000L)
monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null))
monitor.onTaskStart(SparkListenerTaskStart(1, 0, taskInfo("1", 1)))
monitor.onTaskEnd(SparkListenerTaskEnd(1, 0, "foo", Success, taskInfo("1", 1), null))
monitor.onJobEnd(SparkListenerJobEnd(1, clock.getTimeMillis(), JobSucceeded))

assert(monitor.timedOutExecutors(idleDeadline).isEmpty)
}

private def idleDeadline: Long = clock.getTimeMillis() + idleTimeoutMs + 1
private def storageDeadline: Long = clock.getTimeMillis() + storageTimeoutMs + 1
private def shuffleDeadline: Long = clock.getTimeMillis() + shuffleTimeoutMs + 1
Expand Down