-
Notifications
You must be signed in to change notification settings - Fork 28.3k
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.
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
[SPARK-12330] [MESOS] Fix mesos coarse mode cleanup #10319
Changes from all commits
41543e0
0a24530
b9e1c77
9ae106c
f9c15c7
75887c4
1181a05
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -19,11 +19,13 @@ package org.apache.spark.scheduler.cluster.mesos | |
|
||
import java.io.File | ||
import java.util.{Collections, List => JList} | ||
import java.util.concurrent.TimeUnit | ||
import java.util.concurrent.locks.ReentrantLock | ||
|
||
import scala.collection.JavaConverters._ | ||
import scala.collection.mutable.{HashMap, HashSet} | ||
|
||
import com.google.common.base.Stopwatch | ||
import com.google.common.collect.HashBiMap | ||
import org.apache.mesos.{Scheduler => MScheduler, SchedulerDriver} | ||
import org.apache.mesos.Protos.{TaskInfo => MesosTaskInfo, _} | ||
|
@@ -60,6 +62,12 @@ private[spark] class CoarseMesosSchedulerBackend( | |
// Maximum number of cores to acquire (TODO: we'll need more flexible controls here) | ||
val maxCores = conf.get("spark.cores.max", Int.MaxValue.toString).toInt | ||
|
||
private[this] val shutdownTimeoutMS = conf.getTimeAsMs("spark.mesos.coarse.shutdown.ms", "10s") | ||
.ensuring(_ >= 0, "spark.mesos.coarse.shutdown.ms must be >= 0") | ||
|
||
// Synchronization protected by stateLock | ||
private[this] var stopCalled: Boolean = false | ||
|
||
// If shuffle service is enabled, the Spark driver will register with the shuffle service. | ||
// This is for cleaning up shuffle files reliably. | ||
private val shuffleServiceEnabled = conf.getBoolean("spark.shuffle.service.enabled", false) | ||
|
@@ -245,6 +253,13 @@ private[spark] class CoarseMesosSchedulerBackend( | |
*/ | ||
override def resourceOffers(d: SchedulerDriver, offers: JList[Offer]) { | ||
stateLock.synchronized { | ||
if (stopCalled) { | ||
logDebug("Ignoring offers during shutdown") | ||
// Driver should simply return a stopped status on race | ||
// condition between this.stop() and completing here | ||
offers.asScala.map(_.getId).foreach(d.declineOffer) | ||
return | ||
} | ||
val filters = Filters.newBuilder().setRefuseSeconds(5).build() | ||
for (offer <- offers.asScala) { | ||
val offerAttributes = toAttributeMap(offer.getAttributesList) | ||
|
@@ -364,7 +379,29 @@ private[spark] class CoarseMesosSchedulerBackend( | |
} | ||
|
||
override def stop() { | ||
super.stop() | ||
// Make sure we're not launching tasks during shutdown | ||
stateLock.synchronized { | ||
if (stopCalled) { | ||
logWarning("Stop called multiple times, ignoring") | ||
return | ||
} | ||
stopCalled = true | ||
super.stop() | ||
} | ||
// Wait for executors to report done, or else mesosDriver.stop() will forcefully kill them. | ||
// See SPARK-12330 | ||
val stopwatch = new Stopwatch() | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This Stopwatch constructor was deprecated in newer versions of Guava (google/guava@fd0cbc2). In order to work around this issue, I'd like to remove this use of There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sounds good. Sorry, didn't see this note until now, and it looks like this was already fixed in master. |
||
stopwatch.start() | ||
// slaveIdsWithExecutors has no memory barrier, so this is eventually consistent | ||
while (slaveIdsWithExecutors.nonEmpty && | ||
stopwatch.elapsed(TimeUnit.MILLISECONDS) < shutdownTimeoutMS) { | ||
Thread.sleep(100) | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I thought you used to have a check in the end, that if slaveIdsWithExecutors is non empty we print a warning right? I think that's still valuable to print it. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. sure, will make sure that's in There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. added warning if still nonEmpty |
||
if (slaveIdsWithExecutors.nonEmpty) { | ||
logWarning(s"Timed out waiting for ${slaveIdsWithExecutors.size} remaining executors " | ||
+ s"to terminate within $shutdownTimeoutMS ms. This may leave temporary files " | ||
+ "on the mesos nodes.") | ||
} | ||
if (mesosDriver != null) { | ||
mesosDriver.stop() | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you please submit a new PR to add this to the docs?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
no #10319 (comment)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ah, nvm
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
by the way I just realized this config is not properly named. Unfortunately I did not catch this during code reviews and I just pushed a hot fix in master to correct this: c756bda