You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/08/27 18:54:00 UTC
(spark) branch branch-4.x updated: [SPARK-59055][CORE] Report the application hold status to `Spark Master`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new c457fcdde009 [SPARK-59055][CORE] Report the application hold status to `Spark Master`
c457fcdde009 is described below
commit c457fcdde0090a70fa68266cc4396b8d30de3bdb
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Aug 27 11:53:42 2026 -0700
[SPARK-59055][CORE] Report the application hold status to `Spark Master`
### What changes were proposed in this pull request?
This PR lets a driver report its hold status to the standalone Master, so that the Master UI and
`/json/` endpoint can show it. Display only -- the `(hold)` / `(resume)` controls stay on the
driver web UI.
- `SparkContext` calls a new no-op `CoarseGrainedSchedulerBackend.reportExecutorHoldStatus` hook
after initialization and on every hold/resume transition. `StandaloneSchedulerBackend` overrides
it and forwards to `StandaloneAppClient`, which sends the new `ApplicationHoldUpdated` message to
the Master. The status is cached and re-sent on registration and failover.
- The Master mirrors it onto `ApplicationInfo.holdSupported` / `held` (`transient`; a new Master
learns it again from the driver's re-report). The draining count is not pushed: the Master
derives it from the executors it already tracks.
- The Master UI annotates the state column (e.g. `RUNNING (held, draining 2 executors)`), and
`/json/` gains `holdsupported`, `held`, and `draining` fields per application.
`spark.ui.holdEnabled` is honored (an opted-out application is not reported as holdable), and
finished applications are never annotated. No new configuration and no new public API.
### Why are the changes needed?
SPARK-58828 and SPARK-59010 made the hold status visible only per application on the driver. An
operator on the Master page cannot tell which applications are held or still draining. This also
lays the groundwork for offering the controls on the Master UI in a follow-up PR.
### Does this PR introduce _any_ user-facing change?
Yes, additive: the Master UI annotates held applications in the state column, and each application
in `/json/` gains `holdsupported`, `held`, and `draining` fields.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #58346 from dongjoon-hyun/SPARK-59055.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 7ec8aa1eb1f54d19776733c4a5fc8f5e262daff5)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../main/scala/org/apache/spark/SparkContext.scala | 27 ++++++++++++-
.../org/apache/spark/deploy/DeployMessage.scala | 10 +++++
.../org/apache/spark/deploy/JsonProtocol.scala | 6 +++
.../spark/deploy/client/StandaloneAppClient.scala | 29 ++++++++++++++
.../spark/deploy/master/ApplicationInfo.scala | 41 ++++++++++++++++++++
.../org/apache/spark/deploy/master/Master.scala | 10 +++++
.../spark/deploy/master/ui/ApplicationPage.scala | 2 +-
.../apache/spark/deploy/master/ui/MasterPage.scala | 2 +-
.../cluster/CoarseGrainedSchedulerBackend.scala | 7 ++++
.../cluster/StandaloneSchedulerBackend.scala | 4 ++
.../apache/spark/deploy/JsonProtocolSuite.scala | 16 +++++++-
.../deploy/StandaloneDynamicAllocationSuite.scala | 45 ++++++++++++++++++++++
.../spark/deploy/client/AppClientSuite.scala | 37 ++++++++++++++++++
.../apache/spark/deploy/master/MasterSuite.scala | 34 ++++++++++++++++
.../master/ui/ReadOnlyMasterWebUISuite.scala | 22 +++++++++++
docs/spark-standalone.md | 12 ++++++
16 files changed, 299 insertions(+), 5 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 438c036a359a..04d04a1fd382 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -715,6 +715,10 @@ class SparkContext(config: SparkConf) extends Logging {
postEnvironmentUpdate()
postApplicationStart()
+ // Advertise whether this application can be held, now that the shuffle driver components,
+ // which decide it, are up.
+ reportExecutorHoldStatus()
+
// After application started, attach handlers to started server and start handler.
_ui.foreach(_.attachAllHandlers())
// Attach the driver metrics servlet handler to the web ui after the metrics system is started.
@@ -2143,7 +2147,7 @@ class SparkContext(config: SparkConf) extends Logging {
*/
@DeveloperApi
def holdExecutors(): Boolean = {
- schedulerBackend match {
+ val acknowledged = schedulerBackend match {
case cg: CoarseGrainedSchedulerBackend if cg.supportsExecutorHold =>
require(executorHoldSupported,
s"holdExecutors() requires ${DECOMMISSION_ENABLED.key} and either " +
@@ -2203,6 +2207,8 @@ class SparkContext(config: SparkConf) extends Logging {
logWarning("Holding executors is not supported by current scheduler.")
false
}
+ reportExecutorHoldStatus()
+ acknowledged
}
// Gracefully decommission all the current executors of a held application and let the
@@ -2249,7 +2255,7 @@ class SparkContext(config: SparkConf) extends Logging {
*/
@DeveloperApi
def resumeExecutors(): Boolean = {
- schedulerBackend match {
+ val acknowledged = schedulerBackend match {
case cg: CoarseGrainedSchedulerBackend =>
synchronized {
if (!_executorsHeld) {
@@ -2308,6 +2314,23 @@ class SparkContext(config: SparkConf) extends Logging {
logWarning("Resuming executors is not supported by current scheduler.")
false
}
+ reportExecutorHoldStatus()
+ acknowledged
+ }
+
+ /**
+ * Tell the cluster manager whether this application can be held and whether it currently is,
+ * so that it can show the hold status on its own UI. Called once the context is fully started
+ * -- `executorHoldSupported` reads the shuffle driver components, which are initialized late
+ * -- and again after every transition. Synchronized so that concurrent transitions cannot
+ * deliver their reports inverted: whichever call serializes last reports the current state.
+ */
+ private def reportExecutorHoldStatus(): Unit = synchronized {
+ schedulerBackend match {
+ case cg: CoarseGrainedSchedulerBackend =>
+ cg.reportExecutorHoldStatus(executorHoldSupported, _executorsHeld)
+ case _ =>
+ }
}
/** The version of Spark on which this application is running. */
diff --git a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
index 12e031711aa2..8ad8efff4acd 100644
--- a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
@@ -201,6 +201,12 @@ private[deploy] object DeployMessages {
case class KillExecutors(appId: String, executorIds: Seq[String])
+ // Whether this application can be held and whether it currently is. Pushed on every
+ // transition and re-sent on failover, since a new Master starts without it. The number of
+ // executors still draining is not carried here: the Master already tracks the executors of
+ // the application and derives it from them.
+ case class ApplicationHoldUpdated(appId: String, supported: Boolean, held: Boolean)
+
// Master to AppClient
case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
@@ -251,6 +257,10 @@ private[deploy] object DeployMessages {
case object StopAppClient
+ // Reports the hold status of the application to the Master, and caches it in the endpoint so
+ // that it can be re-sent on failover.
+ case class ReportApplicationHold(supported: Boolean, held: Boolean)
+
// Master to Worker & AppClient
case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
diff --git a/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
index 2a3fd0d004e1..1119f47094d6 100644
--- a/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
@@ -95,6 +95,9 @@ private[deploy] object JsonProtocol {
* `resourcesperexecutor` minimal resources required to each executor
* `submitdate` time in Date that the application is submitted
* `state` state of the application, see [[ApplicationState]]
+ * `holdsupported` whether the driver reported that it can be held
+ * `held` whether the application is currently held; always false once it finishes
+ * `draining` the number of executors still draining while the application is held
* `duration` time in milliseconds that the application has been running
* For compatibility also returns the deprecated `memoryperslave` & `resourcesperslave` fields.
*/
@@ -112,6 +115,9 @@ private[deploy] object JsonProtocol {
.toList.map(writeResourceRequirement)) ~
("submitdate" -> obj.submitDate.toString) ~
("state" -> obj.state.toString) ~
+ ("holdsupported" -> obj.holdSupported) ~
+ ("held" -> obj.isHeld) ~
+ ("draining" -> obj.numDrainingExecutors) ~
("duration" -> obj.duration)
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
index 2d742b31f99c..1b85ce06eced 100644
--- a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
@@ -71,6 +71,9 @@ private[spark] class StandaloneAppClient(
private val alreadyDead = new AtomicBoolean(false)
private val registerMasterFutures = new AtomicReference[Array[JFuture[_]]]
private val registrationRetryTimer = new AtomicReference[JScheduledFuture[_]]
+ // The hold status last reported to the Master, re-sent on failover since a new Master
+ // starts without it. None until the driver has reported one.
+ private var holdStatus: Option[(Boolean, Boolean)] = None
// A thread pool for registering with masters. Because registering with a master is a blocking
// action, this thread pool must be able to create "masterRpcAddresses.size" threads at the same
@@ -169,6 +172,7 @@ private[spark] class StandaloneAppClient(
registered.set(true)
master = Some(masterRef)
listener.connected(appId.get)
+ sendHoldStatus()
case ApplicationRemoved(message) =>
markDead("Master removed our application: %s".format(message))
@@ -204,6 +208,18 @@ private[spark] class StandaloneAppClient(
master = Some(masterRef)
alreadyDisconnected = false
masterRef.send(MasterChangeAcknowledged(appId.get))
+ // The new Master recovered the application without its hold status, so report it again.
+ sendHoldStatus()
+
+ case ReportApplicationHold(supported, held) =>
+ holdStatus = Some((supported, held))
+ sendHoldStatus()
+ }
+
+ /** Report the last known hold status, if any, to the current Master. */
+ private def sendHoldStatus(): Unit = holdStatus.foreach { case (supported, held) =>
+ // Dropped while no Master is known; re-sent from `RegisteredApplication`/`MasterChanged`.
+ sendToMaster(ApplicationHoldUpdated(appId.get, supported, held))
}
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
@@ -330,6 +346,19 @@ private[spark] class StandaloneAppClient(
}
}
+ /**
+ * Report to the Master whether this application can be held and whether it currently is, so
+ * that the Master UI can show the hold status of the application. The status is cached and
+ * re-sent on failover, so a report made before the registration completes is not lost.
+ */
+ def reportHoldStatus(supported: Boolean, held: Boolean): Unit = {
+ if (endpoint.get != null) {
+ endpoint.get.send(ReportApplicationHold(supported, held))
+ } else {
+ logWarning("Attempted to report the hold status before driver fully initialized.")
+ }
+ }
+
/**
* Kill the given list of executors through the Master.
* @return whether the kill request is acknowledged.
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala
index e66933b84af5..0c329e96328c 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala
@@ -44,6 +44,11 @@ private[spark] class ApplicationInfo(
@transient var endTime: Long = _
@transient var appSource: ApplicationSource = _
+ // Hold state reported by the driver. Transient, because it belongs to the running driver and
+ // not to the recovered application: a new Master learns it again when the driver re-registers.
+ @transient var holdSupported: Boolean = _
+ @transient var held: Boolean = _
+
@transient private var executorsPerResourceProfileId: mutable.HashMap[Int, mutable.Set[Int]] = _
@transient private var targetNumExecutorsPerResourceProfileId: mutable.HashMap[Int, Int] = _
@transient private var rpIdToResourceProfile: mutable.HashMap[Int, ResourceProfile] = _
@@ -63,6 +68,8 @@ private[spark] class ApplicationInfo(
executors = new mutable.HashMap[Int, ExecutorDesc]
coresGranted = 0
endTime = -1L
+ holdSupported = false
+ held = false
appSource = new ApplicationSource(this)
nextExecutorId = 0
removedExecutors = new ArrayBuffer[ExecutorDesc]
@@ -197,6 +204,40 @@ private[spark] class ApplicationInfo(
targetNumExecutorsPerResourceProfileId.values.sum
}
+ /**
+ * Whether the application is held right now, per the last report from its running driver. An
+ * application that did not report the hold as supported is not considered held -- its driver
+ * UI offers no control to change it -- and neither is a finished application, whose driver is
+ * gone, so the last reported hold is stale.
+ */
+ private[deploy] def isHeld: Boolean = held && holdSupported && !isFinished
+
+ /**
+ * The number of executors that have not exited yet while the application is held. They are
+ * still draining their running tasks; the hold is complete once this reaches zero. Counting
+ * `executors` relies on `DECOMMISSIONED` not being a finished `ExecutorState`: the Master
+ * keeps a decommissioning executor in the map until it actually exits, which is what makes
+ * this count agree with the driver's own `draining` from its `/holdstatus` endpoint.
+ */
+ private[deploy] def numDrainingExecutors: Int = if (isHeld) executors.size else 0
+
+ /**
+ * The application state, annotated with the hold reported by its driver, for example
+ * `RUNNING (held, draining 2 executors)`.
+ */
+ private[deploy] def stateText: String = {
+ if (!isHeld) {
+ state.toString
+ } else {
+ val draining = numDrainingExecutors
+ if (draining == 0) {
+ s"$state (held)"
+ } else {
+ s"$state (held, draining $draining executor${if (draining > 1) "s" else ""})"
+ }
+ }
+ }
+
def duration: Long = {
if (endTime != -1) {
endTime - startTime
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
index e95c4bd8c622..ad42aa5a5e1b 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -393,6 +393,16 @@ private[deploy] class Master(
log" ${MDC(LogKeys.APP_ID, applicationId)}")
idToApp.get(applicationId).foreach(finishApplication)
+ case ApplicationHoldUpdated(appId, supported, held) =>
+ idToApp.get(appId) match {
+ case Some(app) =>
+ app.holdSupported = supported
+ app.held = held
+ case None =>
+ logWarning(log"Got hold status for unknown application " +
+ log"${MDC(LogKeys.APP_ID, appId)}")
+ }
+
case CheckForWorkerTimeOut =>
timeOutDeadWorkers()
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
index 3de530f1d252..095eaa3afe68 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/ApplicationPage.scala
@@ -89,7 +89,7 @@ private[ui] class ApplicationPage(parent: MasterWebUI) extends WebUIPage("app")
</li>
<li><strong>Submit Date:</strong> {UIUtils.formatDate(app.submitDate)}</li>
<li><strong>Duration:</strong> {UIUtils.formatDuration(app.duration)}</li>
- <li><strong>State:</strong> {app.state}</li>
+ <li><strong>State:</strong> {app.stateText}</li>
{
if (!app.isFinished) {
if (app.desc.appUiUrl.isBlank()) {
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
index 2f38b6fcfb4f..af195adcae54 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterPage.scala
@@ -348,7 +348,7 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
</td>
<td>{UIUtils.formatDate(app.submitDate)}</td>
<td>{app.desc.user}</td>
- <td>{app.state.toString}</td>
+ <td>{app.stateText}</td>
<td sorttable_customkey={app.duration.toString}>
{UIUtils.formatDuration(app.duration)}
</td>
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
index 1a2d72b9d741..1b5cd70cef90 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -809,6 +809,13 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
executorsHeld = held
}
+ /**
+ * Report the hold status of the application to the cluster manager, so that it can show the
+ * status on its own UI. Called once the application is fully started and again on every
+ * transition. Ignored by default: only Standalone renders it today.
+ */
+ private[spark] def reportExecutorHoldStatus(supported: Boolean, held: Boolean): Unit = {}
+
/** Whether an executor total was ever explicitly requested. Visible for testing only. */
private[spark] def hasExplicitExecutorRequests: Boolean = synchronized {
explicitExecutorRequest
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
index 751c4def7c92..aebe364c9293 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
@@ -252,6 +252,10 @@ private[spark] class StandaloneSchedulerBackend(
// executors can be held gracefully.
private[spark] override def supportsExecutorHold: Boolean = true
+ private[spark] override def reportExecutorHoldStatus(supported: Boolean, held: Boolean): Unit = {
+ Option(client).foreach(_.reportHoldStatus(supported, held))
+ }
+
/**
* Kill the given list of executors through the Master.
* @return whether the kill request is acknowledged.
diff --git a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
index 6d2c663a2588..ac839103aef5 100644
--- a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
@@ -39,6 +39,18 @@ class JsonProtocolSuite extends SparkFunSuite with JsonTestUtils {
assertValidDataInJson(output, JsonMethods.parse(JsonConstants.appInfoJsonStr))
}
+ test("SPARK-59055: writeApplicationInfo with the hold status") {
+ val appInfo = createAppInfo()
+ appInfo.holdSupported = true
+ appInfo.held = true
+ val output = JsonProtocol.writeApplicationInfo(appInfo)
+ assertValidJson(output)
+ assert(output \ "holdsupported" === JBool(true))
+ assert(output \ "held" === JBool(true))
+ // The application has no executor left, so its hold is complete.
+ assert(output \ "draining" === JInt(0))
+ }
+
test("writeWorkerInfo") {
val output = JsonProtocol.writeWorkerInfo(createWorkerInfo())
assertValidJson(output)
@@ -191,7 +203,9 @@ object JsonConstants {
|"resourcesperslave":[{"name":"fpga",
|"amount":3},{"name":"gpu","amount":3}],
|"submitdate":"%s",
- |"state":"WAITING","duration":%d}
+ |"state":"WAITING",
+ |"holdsupported":false,"held":false,"draining":0,
+ |"duration":%d}
""".format(System.getProperty("user.name", "<unknown>"),
submitDate.toString, currTimeInMillis - appInfoStartTime).stripMargin
diff --git a/core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala b/core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala
index 90ef0aa510c2..4614435af44c 100644
--- a/core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/StandaloneDynamicAllocationSuite.scala
@@ -496,6 +496,51 @@ class StandaloneDynamicAllocationSuite
}
}
+ test("SPARK-59055: SparkContext reports the hold status of the application to the Master") {
+ sc = new SparkContext(appConf
+ .set(config.SHUFFLE_SERVICE_ENABLED, true)
+ .set(config.DECOMMISSION_ENABLED, true))
+ // The report from the end of the SparkContext constructor.
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().length === 1)
+ assert(getApplications().head.holdSupported)
+ }
+ assert(!getApplications().head.held)
+
+ assert(sc.holdExecutors())
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().head.held)
+ }
+ // The faked executors never exit, so all of them are still counted as draining.
+ assert(getApplications().head.numDrainingExecutors === 2)
+
+ assert(sc.resumeExecutors())
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(!getApplications().head.held)
+ }
+ }
+
+ test("SPARK-59055: spark.ui.holdEnabled=false does not suppress the hold status") {
+ sc = new SparkContext(appConf
+ .set(config.SHUFFLE_SERVICE_ENABLED, true)
+ .set(config.DECOMMISSION_ENABLED, true)
+ .set(config.UI.UI_HOLD_ENABLED, false))
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().length === 1)
+ }
+
+ // spark.ui.holdEnabled gates the controls on the driver UI, not the capability itself:
+ // holdExecutors() is a developer API that the config does not touch, so a hold made with
+ // the config off must still be visible on the Master.
+ assert(sc.holdExecutors())
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().head.held)
+ }
+ assert(getApplications().head.holdSupported)
+ assert(getApplications().head.isHeld)
+ assert(getApplications().head.numDrainingExecutors === 2)
+ }
+
test("executor registration on a excluded host must fail") {
// The context isn't really used by the test, but it helps with creating a test scheduler,
// since CoarseGrainedSchedulerBackend makes a lot of calls to the context instance.
diff --git a/core/src/test/scala/org/apache/spark/deploy/client/AppClientSuite.scala b/core/src/test/scala/org/apache/spark/deploy/client/AppClientSuite.scala
index 877aee47cd65..4147535eb0db 100644
--- a/core/src/test/scala/org/apache/spark/deploy/client/AppClientSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/client/AppClientSuite.scala
@@ -217,6 +217,43 @@ class AppClientSuite
}
}
+ test("SPARK-59055: report the hold status of the application to the Master") {
+ Utils.tryWithResource(new AppClientInst(masterRpcEnv.address.toSparkURL)) { ci =>
+ ci.client.start()
+
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().length === 1, "master should have 1 registered app")
+ }
+
+ // The Master knows nothing about the hold until the driver reports it.
+ val app = getApplications().head
+ assert(!app.holdSupported && !app.held && app.numDrainingExecutors === 0)
+
+ ci.client.reportHoldStatus(supported = true, held = false)
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().head.holdSupported)
+ }
+ assert(!getApplications().head.held)
+
+ ci.client.reportHoldStatus(supported = true, held = true)
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().head.held)
+ }
+
+ ci.client.reportHoldStatus(supported = true, held = false)
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(!getApplications().head.held)
+ }
+
+ // Issue stop command for Client to disconnect from Master
+ ci.client.stop()
+
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().isEmpty, "master should have 0 registered apps")
+ }
+ }
+ }
+
test("request from AppClient before initialized with master") {
Utils.tryWithResource(new AppClientInst(masterRpcEnv.address.toSparkURL)) { ci =>
diff --git a/core/src/test/scala/org/apache/spark/deploy/master/MasterSuite.scala b/core/src/test/scala/org/apache/spark/deploy/master/MasterSuite.scala
index c138d1142a47..2b68f3db1b44 100644
--- a/core/src/test/scala/org/apache/spark/deploy/master/MasterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/master/MasterSuite.scala
@@ -250,6 +250,40 @@ class MasterSuite extends MasterSuiteBase {
assert(master.invokePrivate(_createApplication(desc, null)).id === "spark-45756")
}
+ test("SPARK-59055: The executors of a held application are counted as draining") {
+ val appInfo = makeAppInfo(1024)
+ val worker = DeployTestUtils.createWorkerInfo()
+ appInfo.addExecutor(worker, 1, 1024, Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)
+ appInfo.addExecutor(worker, 1, 1024, Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)
+
+ // The executors of an application that is not held are not draining.
+ assert(appInfo.numDrainingExecutors === 0)
+
+ // A hold that the driver did not report as supported is not treated as held.
+ appInfo.held = true
+ assert(appInfo.numDrainingExecutors === 0)
+
+ // While held, an executor that has not exited yet is still draining its running tasks.
+ appInfo.holdSupported = true
+ assert(appInfo.numDrainingExecutors === 2)
+
+ // The hold is complete once the last executor is gone.
+ appInfo.executors.values.toSeq.foreach(appInfo.removeExecutor)
+ assert(appInfo.numDrainingExecutors === 0)
+ assert(appInfo.isHeld)
+ assert(appInfo.stateText === "WAITING (held)")
+
+ // A single draining executor is reported in the singular.
+ appInfo.addExecutor(worker, 1, 1024, Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID)
+ assert(appInfo.stateText === "WAITING (held, draining 1 executor)")
+
+ // A finished application is never held, even though the Master keeps its executors for the
+ // UI: its driver is gone, so the last reported hold is stale.
+ appInfo.markFinished(ApplicationState.FINISHED)
+ assert(!appInfo.isHeld)
+ assert(appInfo.numDrainingExecutors === 0)
+ }
+
test("SPARK-57451: Allows REST server and spark.authenticate.secret to be enabled together") {
val conf = new SparkConf()
.set(MASTER_REST_SERVER_ENABLED, true)
diff --git a/core/src/test/scala/org/apache/spark/deploy/master/ui/ReadOnlyMasterWebUISuite.scala b/core/src/test/scala/org/apache/spark/deploy/master/ui/ReadOnlyMasterWebUISuite.scala
index 2679349bfe02..66661aa1cd6c 100644
--- a/core/src/test/scala/org/apache/spark/deploy/master/ui/ReadOnlyMasterWebUISuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/master/ui/ReadOnlyMasterWebUISuite.scala
@@ -31,6 +31,7 @@ import org.apache.spark.deploy.master.ui.MasterWebUISuite._
import org.apache.spark.internal.config.DECOMMISSION_ENABLED
import org.apache.spark.internal.config.UI.MASTER_UI_VISIBLE_ENV_VAR_PREFIXES
import org.apache.spark.internal.config.UI.UI_KILL_ENABLED
+import org.apache.spark.resource.ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID
import org.apache.spark.rpc.{RpcEndpointRef, RpcEnv}
import org.apache.spark.util.Utils
@@ -73,6 +74,27 @@ class ReadOnlyMasterWebUISuite extends SparkFunSuite {
}
}
+ test("SPARK-59055: annotate the state of a held application") {
+ val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/"
+ app1.holdSupported = true
+ app1.held = true
+ app1.addExecutor(createWorkerInfo(), 1, 1234, Map.empty, DEFAULT_RESOURCE_PROFILE_ID)
+ app1.addExecutor(createWorkerInfo(), 1, 1234, Map.empty, DEFAULT_RESOURCE_PROFILE_ID)
+ try {
+ var result = Source.fromInputStream(sendHttpRequest(url, "GET", "").getInputStream).mkString
+ assert(result.contains("WAITING (held, draining 2 executors)"))
+
+ // An application that did not report the hold as supported is not annotated.
+ app1.holdSupported = false
+ result = Source.fromInputStream(sendHttpRequest(url, "GET", "").getInputStream).mkString
+ assert(!result.contains("(held"))
+ } finally {
+ app1.holdSupported = false
+ app1.held = false
+ app1.executors.values.toSeq.foreach(app1.removeExecutor)
+ }
+ }
+
test("/app/kill POST method is not allowed") {
val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/kill/"
val body = convPostDataToString(Map(("id", "1"), ("terminate", "true")))
diff --git a/docs/spark-standalone.md b/docs/spark-standalone.md
index b35e74d246ad..125ae01428cc 100644
--- a/docs/spark-standalone.md
+++ b/docs/spark-standalone.md
@@ -765,6 +765,18 @@ In addition, detailed log output for each job is also written to the work direct
To track and review logs across completed applications, [enable event logging and start the History Server](monitoring.html#viewing-after-the-fact).
+## Held Applications
+
+An application that can be held reports to the Master whether it currently is, and the Master web
+UI annotates the application state accordingly, for example `RUNNING (held, draining 2 executors)`.
+An executor that has not exited yet is still finishing its running tasks, and the hold is complete
+once no executor is left. The Master's `/json/` endpoint reports the same in the `holdsupported`,
+`held`, and `draining` fields of each application.
+
+Only applications whose driver reports that it can be held are annotated, per the preconditions
+described in [Web UI](web-ui.html#jobs-tab). Holding and resuming an application is done from its
+own driver web UI.
+
# Running Alongside Hadoop
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org