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 23:58:41 UTC
(spark) branch branch-4.x updated: [SPARK-59061][CORE] Hold and resume an application from the Master UI
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 06587044b6f4 [SPARK-59061][CORE] Hold and resume an application from the Master UI
06587044b6f4 is described below
commit 06587044b6f4038be77b35c406c36588914a63af
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Aug 27 16:58:23 2026 -0700
[SPARK-59061][CORE] Hold and resume an application from the Master UI
### What changes were proposed in this pull request?
This PR adds `(hold)` and `(resume)` controls to the standalone Master web UI, next to the existing
`(kill)` link, building on the hold status reported by SPARK-59055:
- https://github.com/apache/spark/pull/58346
The Master does not hold the application itself; it forwards the request to the driver, which owns
the hold (`SparkContext.holdExecutors()` / `resumeExecutors()`, SPARK-58828). Holding it on the
Master side instead would bypass the driver's registration guard, be undone by the next
`requestTotalExecutors`, and disagree with the driver's own `/holdstatus` endpoint.
```mermaid
sequenceDiagram
participant W as Master Web UI
participant M as Master
participant C as StandaloneAppClient
participant B as StandaloneSchedulerBackend
participant S as SparkContext
W->>M: POST /app/hold or /app/resume<br/>RequestApplicationHold (send)
M->>C: SetApplicationHold (ask, not awaited on the dispatcher)
C->>B: listener.holdApplication
Note over B,S: dedicated single thread (standalone-hold-resume)
B->>S: holdExecutors() / resumeExecutors()
S-->>B: reportExecutorHoldStatus(supported, held)
B-->>C: reportHoldStatus
C-->>M: ApplicationHoldUpdated (SPARK-59055)
C-->>M: reply whether acknowledged (logged on the Master)
```
- Two new messages: `RequestApplicationHold` (Master UI to Master) and `SetApplicationHold`
(Master to app client, replying whether the request was acknowledged). The resulting state comes
back through the existing `ApplicationHoldUpdated` push, unchanged from SPARK-59055.
- The controls follow the existing `(kill)` pattern: a `POST`-only form, registered under
`spark.ui.holdEnabled` the way the kill handlers are registered under `spark.ui.killEnabled`, and
requiring modify permissions. The generic `confirm-link` handler in `webui.js` now submits the
form it is wrapped in, so it can guard a `POST` control and not only a navigation.
- `spark.ui.holdEnabled` on the application is honored as a static policy: it travels with the
registration as a new `ApplicationDescription.holdEnabled` field, following the
`initialExecutorLimit` precedent. The Master offers no controls for an application that disabled
it and rejects its hold requests without the round trip to the driver, which also rejects them as
a second line of defense. The dynamic hold status reported by SPARK-59055 is untouched, so such
an application stays fully visible -- state annotation, `held`, and `draining` -- when it is held
programmatically; only the controls are withheld. Since the description is persisted for
recovery, a failed-over Master knows the policy before the driver re-reports.
Nothing blocks on the driver: the Master `send`s rather than asks from the Jetty thread, the Master
does not wait on its ask to the driver, and the driver serves the request on a dedicated
single-thread pool rather than on an RPC endpoint thread. Like the kill links, the request itself
gets no inline feedback: the UI reflects the resulting state, and the outcome of the request is
logged on the Master.
No new configuration and no new public API.
### Why are the changes needed?
SPARK-59055 made the hold status visible on the Master page, but acting on it still requires
reaching each driver's own web UI. In a cluster where drivers are not directly reachable, the Master
page is the operator's only console, and it is already where applications are killed. Holding is the
graceful alternative to killing -- the application gives back its executors, keeps its driver and
the shuffle output already written, and can be resumed later -- so it belongs next to `(kill)`.
### Does this PR introduce _any_ user-facing change?
Yes, additive. Running applications that report themselves as holdable gain a `(hold)` link, and a
`(resume)` link while held, on the Master web UI. No existing endpoint or field changes its
meaning.
<img width="1471" height="882" alt="Screenshot 2026-08-27 at 13 01 45" src="https://github.com/user-attachments/assets/4c8d12c1-f474-476b-9c33-5674494d6df8" />
<img width="1471" height="882" alt="Screenshot 2026-08-27 at 13 02 03" src="https://github.com/user-attachments/assets/6218c209-2f6e-4277-9213-28241218f253" />
<img width="1471" height="882" alt="Screenshot 2026-08-27 at 13 02 10" src="https://github.com/user-attachments/assets/5cbfb1d1-6e2c-4c6d-a3ca-c7ba2ca125e3" />
### How was this patch tested?
Pass the CIs with new unit tests:
- `MasterWebUISuite`: `/app/hold/` and `/app/resume/` forward the request to the Master; the page
offers the hold or the resume control but never both, and offers neither for an application that
disabled holding while still annotating its state.
- `ReadOnlyMasterWebUISuite`: both endpoints are absent when `spark.ui.holdEnabled` is false on
the Master.
- `AppClientSuite`: a hold and a resume requested from a real Master reach the driver's listener,
and a hold request for an application that disabled holding is rejected by the Master without
reaching the driver.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #58361 from dongjoon-hyun/SPARK-59061.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 2cc6ec7641002f9ae8de989f334d85a1f3e3c8da)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../resources/org/apache/spark/ui/static/webui.js | 6 +-
.../spark/deploy/ApplicationDescription.scala | 3 +
.../org/apache/spark/deploy/DeployMessage.scala | 5 ++
.../spark/deploy/client/StandaloneAppClient.scala | 9 +++
.../client/StandaloneAppClientListener.scala | 14 ++++-
.../org/apache/spark/deploy/master/Master.scala | 41 ++++++++++++-
.../apache/spark/deploy/master/ui/MasterPage.scala | 42 ++++++++++++-
.../spark/deploy/master/ui/MasterWebUI.scala | 8 +++
.../cluster/StandaloneSchedulerBackend.scala | 36 ++++++++++-
.../spark/deploy/client/AppClientSuite.scala | 69 +++++++++++++++++++++-
.../spark/deploy/master/ui/MasterWebUISuite.scala | 64 +++++++++++++++++++-
.../master/ui/ReadOnlyMasterWebUISuite.scala | 14 +++++
docs/spark-standalone.md | 13 +++-
13 files changed, 311 insertions(+), 13 deletions(-)
diff --git a/core/src/main/resources/org/apache/spark/ui/static/webui.js b/core/src/main/resources/org/apache/spark/ui/static/webui.js
index aab41a6a6717..58b530919a77 100644
--- a/core/src/main/resources/org/apache/spark/ui/static/webui.js
+++ b/core/src/main/resources/org/apache/spark/ui/static/webui.js
@@ -131,10 +131,14 @@ $(function() {
}
});
- // generic links guarded by a confirmation prompt before navigating
+ // generic links guarded by a confirmation prompt before navigating, or before submitting the
+ // form they are wrapped in
$(document).on("click", "a.confirm-link[data-confirm-message]", function(e) {
if (!window.confirm($(this).data("confirm-message"))) {
e.preventDefault();
+ } else if ($(this).closest("form").length > 0) {
+ e.preventDefault();
+ $(this).closest("form").submit();
}
});
diff --git a/core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala b/core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala
index 67d0d851b60f..15b882025473 100644
--- a/core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala
@@ -33,6 +33,9 @@ private[spark] case class ApplicationDescription(
// number of executors this application wants to start with,
// only used if dynamic allocation is enabled
initialExecutorLimit: Option[Int] = None,
+ // whether a web UI may offer the hold and resume controls for this application
+ // (spark.ui.holdEnabled on the application)
+ holdEnabled: Boolean = true,
user: String = System.getProperty("user.name", "<unknown>")) {
def memoryPerExecutorMB: Int = defaultProfile.getExecutorMemory.map(_.toInt).getOrElse(1024)
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 8ad8efff4acd..28001f06d46f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
@@ -224,6 +224,9 @@ private[deploy] object DeployMessages {
case class WorkerRemoved(id: String, host: String, message: String)
+ // Hold or resume the whole application, replying whether the request was acknowledged.
+ case class SetApplicationHold(hold: Boolean)
+
// DriverClient <-> Master
case class RequestSubmitDriver(driverDescription: DriverDescription) extends DeployMessage
@@ -269,6 +272,8 @@ private[deploy] object DeployMessages {
case object RequestMasterState
+ case class RequestApplicationHold(appId: String, hold: Boolean)
+
// Master to MasterWebUI
case class MasterStateResponse(
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 1b85ce06eced..05aecabeb615 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
@@ -244,6 +244,15 @@ private[spark] class StandaloneAppClient(
logWarning("Attempted to kill executors before registering with Master.")
context.reply(false)
}
+
+ case SetApplicationHold(hold) =>
+ // Holding drains the executors and talks to the cluster manager, so the listener does
+ // the work off this thread and the reply follows the future it returns.
+ listener.holdApplication(hold).andThen {
+ case Success(acknowledged) => context.reply(acknowledged)
+ case Failure(_: InterruptedException) => // Cancelled
+ case Failure(NonFatal(t)) => context.sendFailure(t)
+ }(ThreadUtils.sameThread)
}
private def askAndReplyAsync[T](
diff --git a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClientListener.scala b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClientListener.scala
index 76970ac9829c..000a46daa8d4 100644
--- a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClientListener.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClientListener.scala
@@ -17,12 +17,15 @@
package org.apache.spark.deploy.client
+import scala.concurrent.Future
+
import org.apache.spark.scheduler.ExecutorDecommissionInfo
/**
- * Callbacks invoked by deploy client when various events happen. There are currently five events:
+ * Callbacks invoked by deploy client when various events happen. There are currently six events:
* connecting to the cluster, disconnecting, being given an executor, having an executor removed
- * (either due to failure or due to revocation), and having a worker removed.
+ * (either due to failure or due to revocation), having a worker removed, and being asked by the
+ * Master to hold or resume the application.
*
* Users of this API should *not* block inside the callback methods.
*/
@@ -44,4 +47,11 @@ private[spark] trait StandaloneAppClientListener {
def executorDecommissioned(fullId: String, decommissionInfo: ExecutorDecommissionInfo): Unit
def workerRemoved(workerId: String, host: String, message: String): Unit
+
+ /**
+ * Hold or resume the whole application on behalf of the Master. Holding drains the executors
+ * and talks to the cluster manager, so the work is done asynchronously and the returned future
+ * completes with whether the request was acknowledged.
+ */
+ def holdApplication(hold: Boolean): Future[Boolean]
}
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 ad42aa5a5e1b..1c3188b91407 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
@@ -23,7 +23,7 @@ import java.util.{Date, Locale}
import java.util.concurrent.{ScheduledFuture, TimeUnit}
import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet}
-import scala.util.Random
+import scala.util.{Failure, Random, Success}
import org.apache.spark.{SecurityManager, SparkConf}
import org.apache.spark.deploy.{ApplicationDescription, DriverDescription, ExecutorState, SparkSubmit}
@@ -403,6 +403,9 @@ private[deploy] class Master(
log"${MDC(LogKeys.APP_ID, appId)}")
}
+ case RequestApplicationHold(appId, hold) =>
+ handleApplicationHold(appId, hold)
+
case CheckForWorkerTimeOut =>
timeOutDeadWorkers()
@@ -1231,6 +1234,42 @@ private[deploy] class Master(
}
}
+ /**
+ * Handle a hold or resume request made from the Master UI by forwarding it to the driver,
+ * which owns the hold: see `SparkContext.holdExecutors()`.
+ *
+ * The driver drains its executors before answering, so the ask is not waited on here -- that
+ * would block the dispatcher. The resulting state arrives separately as
+ * [[ApplicationHoldUpdated]] and the UI renders that; like the kill links, the request itself
+ * gets no UI feedback, so the outcome is only logged.
+ */
+ private def handleApplicationHold(appId: String, hold: Boolean): Unit = {
+ val action = if (hold) "hold" else "resume"
+ idToApp.get(appId) match {
+ case Some(app) if !app.desc.holdEnabled =>
+ // The UI offers no control for such an application, so this is a stale or hand-crafted
+ // request. The driver would reject it too; answer here without the round trip.
+ logWarning(log"Ignoring the ${MDC(LogKeys.OPERATION, action)} request for application " +
+ log"${MDC(LogKeys.APP_ID, appId)}, which disabled holding.")
+ case Some(app) if !app.isFinished =>
+ logInfo(log"Requesting application ${MDC(LogKeys.APP_ID, appId)} to " +
+ log"${MDC(LogKeys.OPERATION, action)}.")
+ app.driver.ask[Boolean](SetApplicationHold(hold)).onComplete {
+ case Success(acknowledged) =>
+ if (!acknowledged) {
+ logWarning(log"The ${MDC(LogKeys.OPERATION, action)} request for application " +
+ log"${MDC(LogKeys.APP_ID, appId)} did not take effect, see the driver logs.")
+ }
+ case Failure(t) =>
+ logWarning(log"Failed to ${MDC(LogKeys.OPERATION, action)} application " +
+ log"${MDC(LogKeys.APP_ID, appId)}", t)
+ }(ThreadUtils.sameThread)
+ case _ =>
+ logWarning(log"Ignoring the ${MDC(LogKeys.OPERATION, action)} request for unknown or " +
+ log"finished application ${MDC(LogKeys.APP_ID, appId)}.")
+ }
+ }
+
/**
* Handle a kill request from the given application.
*
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 af195adcae54..637109cec46b 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
@@ -22,7 +22,7 @@ import scala.xml.Node
import jakarta.servlet.http.HttpServletRequest
import org.json4s.JValue
-import org.apache.spark.deploy.DeployMessages.{KillDriverResponse, MasterStateResponse, RequestKillDriver, RequestMasterState}
+import org.apache.spark.deploy.DeployMessages.{KillDriverResponse, MasterStateResponse, RequestApplicationHold, RequestKillDriver, RequestMasterState}
import org.apache.spark.deploy.JsonProtocol
import org.apache.spark.deploy.StandaloneResourceUtils._
import org.apache.spark.deploy.master._
@@ -62,6 +62,27 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
})
}
+ def handleAppHoldRequest(request: HttpServletRequest): Unit = {
+ handleHoldRequest(request, hold = true)
+ }
+
+ def handleAppResumeRequest(request: HttpServletRequest): Unit = {
+ handleHoldRequest(request, hold = false)
+ }
+
+ private def handleHoldRequest(request: HttpServletRequest, hold: Boolean): Unit = {
+ if (parent.holdEnabled &&
+ parent.master.securityMgr.checkModifyPermissions(request.getRemoteUser)) {
+ Option(request.getParameter("id")).foreach { id =>
+ // Sent rather than asked: the driver drains its executors before answering, which takes
+ // far longer than this request should. The outcome is rendered on a later page load.
+ master.send(RequestApplicationHold(id, hold))
+
+ Thread.sleep(100)
+ }
+ }
+ }
+
private def handleKillRequest(request: HttpServletRequest, action: String => Unit): Unit = {
if (parent.killEnabled &&
parent.master.securityMgr.checkModifyPermissions(request.getRemoteUser)) {
@@ -322,9 +343,28 @@ private[ui] class MasterPage(parent: MasterWebUI) extends WebUIPage("") {
class="kill-link float-end">(kill)</a>
</form>
}
+ // Offered only for an application whose driver reported it as holdable and which did not
+ // disable the controls itself through spark.ui.holdEnabled. Rendered before the kill link:
+ // floated links stack right to left, so this keeps the two controls adjacent and leaves the
+ // kill link's left margin between the application id and the pair.
+ val holdLink = if (parent.holdEnabled && app.desc.holdEnabled && app.holdSupported &&
+ !app.isFinished) {
+ val (action, message) = if (app.isHeld) {
+ ("resume", s"Are you sure you want to resume application ${app.id} ?")
+ } else {
+ ("hold", s"Are you sure you want to hold application ${app.id}? All executors will " +
+ "be decommissioned after finishing their running tasks, and cached blocks are lost.")
+ }
+ <form action={s"app/$action/"} method="POST" class="d-inline">
+ <input type="hidden" name="id" value={app.id}/>
+ <a href="#" data-confirm-message={message}
+ class="confirm-link float-end">{s"($action)"}</a>
+ </form>
+ }
<tr>
<td>
<a href={"app/?appId=" + app.id}>{app.id}</a>
+ {holdLink}
{killLink}
</td>
<td>
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
index 00fe3d260cab..343cfc1b6da0 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ui/MasterWebUI.scala
@@ -28,6 +28,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{HOSTS, NUM_REMOVED_WORKERS}
import org.apache.spark.internal.config.DECOMMISSION_ENABLED
import org.apache.spark.internal.config.UI.MASTER_UI_DECOMMISSION_ALLOW_MODE
+import org.apache.spark.internal.config.UI.UI_HOLD_ENABLED
import org.apache.spark.internal.config.UI.UI_KILL_ENABLED
import org.apache.spark.ui.{SparkUI, WebUI}
import org.apache.spark.ui.JettyUtils._
@@ -45,6 +46,7 @@ class MasterWebUI(
val masterEndpointRef = master.self
val killEnabled = master.conf.get(UI_KILL_ENABLED)
+ val holdEnabled = master.conf.get(UI_HOLD_ENABLED)
val decommissionEnabled = master.conf.get(DECOMMISSION_ENABLED)
val decommissionAllowMode = master.conf.get(MASTER_UI_DECOMMISSION_ALLOW_MODE)
@@ -69,6 +71,12 @@ class MasterWebUI(
attachHandler(createRedirectHandler(
"/driver/kill", "/", masterPage.handleDriverKillRequest, httpMethods = Set("POST")))
}
+ if (holdEnabled) {
+ attachHandler(createRedirectHandler(
+ "/app/hold", "/", masterPage.handleAppHoldRequest, httpMethods = Set("POST")))
+ attachHandler(createRedirectHandler(
+ "/app/resume", "/", masterPage.handleAppResumeRequest, httpMethods = Set("POST")))
+ }
if (decommissionEnabled) {
attachHandler(createServletHandler("/workers/kill", new HttpServlet {
override def doPost(req: HttpServletRequest, resp: HttpServletResponse): Unit = {
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 aebe364c9293..4bb1bf89d2ff 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
@@ -21,7 +21,7 @@ import java.util.Locale
import java.util.concurrent.{RejectedExecutionException, Semaphore, TimeUnit}
import java.util.concurrent.atomic.AtomicBoolean
-import scala.concurrent.Future
+import scala.concurrent.{ExecutionContext, Future}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.deploy.{ApplicationDescription, Command}
@@ -69,6 +69,13 @@ private[spark] class StandaloneSchedulerBackend(
ThreadUtils.newDaemonSingleThreadScheduledExecutor("driver-executor-delay-remove-thread")
private val _executorRemoveDelay = conf.get(EXECUTOR_REMOVE_DELAY)
+ // Serves the Master's hold and resume requests off the RPC endpoint threads: they drain the
+ // executors and talk to the Master, and may block up to the RPC ask timeout. A single thread
+ // also serializes concurrent requests. The pool creates its thread on the first request, so an
+ // application that is never held does not pay for one.
+ private val holdRequestContext = ExecutionContext.fromExecutorService(
+ ThreadUtils.newDaemonSingleThreadExecutor("standalone-hold-resume"))
+
override def start(): Unit = {
super.start()
@@ -131,7 +138,8 @@ private[spark] class StandaloneSchedulerBackend(
None
}
val appDesc = ApplicationDescription(sc.appName, maxCores, command,
- webUrl, defaultProfile = defaultProf, sc.eventLogDir, sc.eventLogCodec, initialExecutorLimit)
+ webUrl, defaultProfile = defaultProf, sc.eventLogDir, sc.eventLogCodec, initialExecutorLimit,
+ holdEnabled = conf.get(config.UI.UI_HOLD_ENABLED))
client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
client.start()
launcherBackend.setState(SparkAppHandle.State.SUBMITTED)
@@ -220,6 +228,21 @@ private[spark] class StandaloneSchedulerBackend(
removeWorker(workerId, host, message)
}
+ override def holdApplication(hold: Boolean): Future[Boolean] = {
+ if (!holdControlEnabled) {
+ // The Master offers the controls only for an application reported as holdable, so this is
+ // a stale or hand-crafted request. Reject it rather than acting on it.
+ logWarning(log"Ignoring the hold request from the Master because " +
+ log"${MDC(LogKeys.CONFIG, config.UI.UI_HOLD_ENABLED.key)} is disabled or the " +
+ log"deployment does not support holding.")
+ Future.successful(false)
+ } else {
+ Future {
+ if (hold) sc.holdExecutors() else sc.resumeExecutors()
+ }(holdRequestContext)
+ }
+ }
+
override def sufficientResourcesRegistered(): Boolean = {
totalCoreCount.get() >= totalExpectedCores * minRegisteredRatio
}
@@ -252,6 +275,14 @@ private[spark] class StandaloneSchedulerBackend(
// executors can be held gracefully.
private[spark] override def supportsExecutorHold: Boolean = true
+ /**
+ * Whether the Master may hold this application. `spark.ui.holdEnabled` gates the controls on
+ * the driver UI, and it gates the Master UI's controls the same way: an application that opted
+ * out of being held should not be holdable from either page.
+ */
+ private def holdControlEnabled: Boolean =
+ sc.executorHoldSupported && conf.get(config.UI.UI_HOLD_ENABLED)
+
private[spark] override def reportExecutorHoldStatus(supported: Boolean, held: Boolean): Unit = {
Option(client).foreach(_.reportHoldStatus(supported, held))
}
@@ -288,6 +319,7 @@ private[spark] class StandaloneSchedulerBackend(
if (stopping.compareAndSet(false, true)) {
try {
executorDelayRemoveThread.shutdownNow()
+ holdRequestContext.shutdownNow()
super.stop()
if (client != null) {
client.stop()
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 4147535eb0db..814d68587a83 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
@@ -20,13 +20,15 @@ package org.apache.spark.deploy.client
import java.io.Closeable
import java.util.concurrent.{ConcurrentHashMap, ConcurrentLinkedQueue}
+import scala.concurrent.Future
import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
import org.scalatest.concurrent.{Eventually, ScalaFutures}
import org.apache.spark._
import org.apache.spark.deploy.{ApplicationDescription, Command, DeployTestUtils}
-import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, RequestMasterState, WorkerDecommissioning}
+import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, RequestApplicationHold, RequestMasterState, WorkerDecommissioning}
import org.apache.spark.deploy.master.{ApplicationInfo, Master}
import org.apache.spark.deploy.worker.Worker
import org.apache.spark.internal.{config, Logging}
@@ -254,6 +256,60 @@ class AppClientSuite
}
}
+ test("SPARK-59061: hold and resume an application from 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")
+ }
+ val appId = getApplications().head.id
+
+ // A hold requested from the Master UI reaches the driver.
+ master.self.send(RequestApplicationHold(appId, hold = true))
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(ci.listener.holdRequestList.asScala.toSeq === Seq(true))
+ }
+
+ master.self.send(RequestApplicationHold(appId, hold = false))
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(ci.listener.holdRequestList.asScala.toSeq === Seq(true, false))
+ }
+
+ ci.client.stop()
+
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().isEmpty, "master should have 0 registered apps")
+ }
+ }
+ }
+
+ test("SPARK-59061: reject the hold request of an application that disabled holding") {
+ Utils.tryWithResource(
+ new AppClientInst(masterRpcEnv.address.toSparkURL, holdEnabled = false)) { ci =>
+ ci.client.start()
+
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().length === 1, "master should have 1 registered app")
+ }
+ val app = getApplications().head
+
+ // The Master rejects without forwarding, so the driver never sees the request. The
+ // askSync afterwards is a barrier: the Master processes its messages in order, so once
+ // it answers, the hold request has been handled.
+ master.self.send(RequestApplicationHold(app.id, hold = true))
+ getApplications()
+ Thread.sleep(100)
+ assert(ci.listener.holdRequestList.isEmpty)
+
+ 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 =>
@@ -304,6 +360,7 @@ class AppClientSuite
val execAddedList = new ConcurrentLinkedQueue[String]()
val execRemovedList = new ConcurrentLinkedQueue[String]()
val execDecommissionedMap = new ConcurrentHashMap[String, ExecutorDecommissionInfo]()
+ val holdRequestList = new ConcurrentLinkedQueue[Boolean]()
def connected(id: String): Unit = {
connectedIdList.add(id)
@@ -339,16 +396,22 @@ class AppClientSuite
}
def workerRemoved(workerId: String, host: String, message: String): Unit = {}
+
+ def holdApplication(hold: Boolean): Future[Boolean] = {
+ holdRequestList.add(hold)
+ Future.successful(true)
+ }
}
/** Create AppClient and supporting objects */
- private class AppClientInst(masterUrl: String) extends Closeable {
+ private class AppClientInst(masterUrl: String, holdEnabled: Boolean = true) extends Closeable {
val rpcEnv = RpcEnv.create("spark", Utils.localHostName(), 0, conf, securityManager)
private val cmd = new Command(TestExecutor.getClass.getCanonicalName.stripSuffix("$"),
List(), Map(), Seq(), Seq(), Seq())
private val defaultRp = DeployTestUtils.createDefaultResourceProfile(512)
val desc =
- ApplicationDescription("AppClientSuite", Some(1), cmd, "ignored", defaultRp)
+ ApplicationDescription("AppClientSuite", Some(1), cmd, "ignored", defaultRp,
+ holdEnabled = holdEnabled)
val listener = new AppClientCollector
val client = new StandaloneAppClient(rpcEnv, Array(masterUrl), desc, listener, new SparkConf)
diff --git a/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala b/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala
index 332e7bf9b5e0..53716e980eac 100644
--- a/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/master/ui/MasterWebUISuite.scala
@@ -23,12 +23,13 @@ import java.nio.charset.StandardCharsets
import java.util.Date
import scala.collection.mutable.HashMap
+import scala.io.Source
import jakarta.servlet.http.HttpServletResponse.SC_FORBIDDEN
import org.mockito.Mockito.{mock, times, verify, when}
import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite}
-import org.apache.spark.deploy.DeployMessages.{DecommissionWorkersOnHosts, KillDriverResponse, RequestKillDriver}
+import org.apache.spark.deploy.DeployMessages.{DecommissionWorkersOnHosts, KillDriverResponse, MasterStateResponse, RequestApplicationHold, RequestKillDriver, RequestMasterState}
import org.apache.spark.deploy.DeployTestUtils._
import org.apache.spark.deploy.master._
import org.apache.spark.internal.config.DECOMMISSION_ENABLED
@@ -91,6 +92,67 @@ class MasterWebUISuite extends SparkFunSuite {
verify(masterEndpointRef, times(1)).ask[KillDriverResponse](RequestKillDriver(activeDriverId))
}
+ private def testHoldApplication(action: String, hold: Boolean): Unit = {
+ val appId = s"app-$action"
+ val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/$action/"
+ val conn = sendHttpRequest(url, "POST", convPostDataToString(Map(("id", appId))))
+ conn.getResponseCode
+
+ // Verify that the master was asked to forward the request to the driver of that application
+ verify(masterEndpointRef, times(1)).send(RequestApplicationHold(appId, hold))
+ }
+
+ test("SPARK-59061: hold application") {
+ testHoldApplication("hold", hold = true)
+ }
+
+ test("SPARK-59061: resume application") {
+ testHoldApplication("resume", hold = false)
+ }
+
+ test("SPARK-59061: offer the hold or the resume control, but never both") {
+ val app = new ApplicationInfo(
+ new Date().getTime, "app-hold", createAppDesc(), new Date(), null, Int.MaxValue)
+ val state = new MasterStateResponse(
+ "host", 8080, None, Array.empty, Array(app), Array.empty,
+ Array.empty, Array.empty, RecoveryState.ALIVE)
+ when(masterEndpointRef.askSync[MasterStateResponse](RequestMasterState)).thenReturn(state)
+ val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/"
+ def render(): String =
+ Source.fromInputStream(sendHttpRequest(url, "GET", "").getInputStream).mkString
+
+ // Nothing is offered until the driver reports the application as holdable.
+ assert(!render().contains("app/hold/") && !render().contains("app/resume/"))
+
+ app.holdSupported = true
+ var result = render()
+ assert(result.contains("app/hold/") && !result.contains("app/resume/"))
+
+ app.held = true
+ result = render()
+ assert(result.contains("app/resume/") && !result.contains("app/hold/"))
+ assert(result.contains("(held)"))
+ }
+
+ test("SPARK-59061: offer no control for an application that disabled holding") {
+ val app = new ApplicationInfo(new Date().getTime, "app-opted-out",
+ createAppDesc().copy(holdEnabled = false), new Date(), null, Int.MaxValue)
+ val state = new MasterStateResponse(
+ "host", 8080, None, Array.empty, Array(app), Array.empty,
+ Array.empty, Array.empty, RecoveryState.ALIVE)
+ when(masterEndpointRef.askSync[MasterStateResponse](RequestMasterState)).thenReturn(state)
+ val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/"
+
+ // The application can be held (e.g. programmatically) and its status stays visible, but the
+ // Master UI offers no control for it.
+ app.holdSupported = true
+ app.held = true
+ val result =
+ Source.fromInputStream(sendHttpRequest(url, "GET", "").getInputStream).mkString
+ assert(!result.contains("app/hold/") && !result.contains("app/resume/"))
+ assert(result.contains("(held)"))
+ }
+
private def testKillWorkers(hostnames: Seq[String]): Unit = {
val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/workers/kill/"
val body = convPostDataToString(hostnames.map(("host", _)))
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 66661aa1cd6c..c613faf2b7ae 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
@@ -30,6 +30,7 @@ import org.apache.spark.deploy.master._
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_HOLD_ENABLED
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}
@@ -41,6 +42,7 @@ class ReadOnlyMasterWebUISuite extends SparkFunSuite {
val conf = new SparkConf()
.set(UI_KILL_ENABLED, false)
+ .set(UI_HOLD_ENABLED, false)
.set(DECOMMISSION_ENABLED, false)
.set(MASTER_UI_VISIBLE_ENV_VAR_PREFIXES.key, "SPARK_SCALA_")
val securityMgr = new SecurityManager(conf)
@@ -107,6 +109,18 @@ class ReadOnlyMasterWebUISuite extends SparkFunSuite {
assert(sendHttpRequest(url, "POST", body).getResponseCode === SC_METHOD_NOT_ALLOWED)
}
+ test("SPARK-59061: /app/hold POST method is not allowed") {
+ val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/hold/"
+ val body = convPostDataToString(Map(("id", "app1")))
+ assert(sendHttpRequest(url, "POST", body).getResponseCode === SC_METHOD_NOT_ALLOWED)
+ }
+
+ test("SPARK-59061: /app/resume POST method is not allowed") {
+ val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/app/resume/"
+ val body = convPostDataToString(Map(("id", "app1")))
+ assert(sendHttpRequest(url, "POST", body).getResponseCode === SC_METHOD_NOT_ALLOWED)
+ }
+
test("/workers/kill POST method is not allowed") {
val hostnames = Seq(s"${Utils.localHostNameForURI()}")
val url = s"http://${Utils.localHostNameForURI()}:${masterWebUI.boundPort}/workers/kill/"
diff --git a/docs/spark-standalone.md b/docs/spark-standalone.md
index 125ae01428cc..ca31146be853 100644
--- a/docs/spark-standalone.md
+++ b/docs/spark-standalone.md
@@ -774,8 +774,17 @@ once no executor is left. The Master's `/json/` endpoint reports the same in the
`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.
+described in [Web UI](web-ui.html#jobs-tab).
+
+Such an application also gets a **(hold)** link next to its **(kill)** link, and a **(resume)**
+link while it is held. Holding stops requesting new executors for the application and gracefully
+decommissions the running ones; the application keeps its driver and the shuffle output already
+written, while cached blocks are lost and recomputed after resuming. Holding and resuming require
+modify permissions, like killing. The links are gated by `spark.ui.holdEnabled` twice: on the
+Master, where setting it to false hides them for every application, and on each application,
+whose own setting travels with its registration -- an application that disabled it gets no links
+and its hold requests are rejected, while its hold status stays visible. The same controls remain
+available on the 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