This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 3e4995134437 [SPARK-59916][CORE] Support `hold` and `resume` in REST
Submission API
3e4995134437 is described below
commit 3e4995134437f1931b54b309bfcf4a2983c58650
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Sep 30 21:59:17 2026 -0700
[SPARK-59916][CORE] Support `hold` and `resume` in REST Submission API
### What changes were proposed in this pull request?
This PR adds `hold` and `resume` actions to the REST Submission API of the
Spark Standalone Master:
- `POST /v1/submissions/hold/<app-id>`
- `POST /v1/submissions/resume/<app-id>`
They do what the Master web UI's **Hold** and **Resume** buttons
(SPARK-59061) do, for scripts:
- https://github.com/apache/spark/pull/58361
Unlike `kill` and `status`, these actions take an application ID rather
than a submission ID, so they also work for applications submitted in client
mode. This is the first REST Submission API action keyed by an application ID.
A submission ID cannot be used here: the Master does not link a driver to the
application it registers, and a client-mode application has no submission ID.
```mermaid
sequenceDiagram
participant R as REST client
participant S as StandaloneRestServer
participant M as Master
participant C as StandaloneAppClient (driver)
R->>S: POST /v1/submissions/hold/<app-id>
S->>M: RequestApplicationHold (ask)
Note over M: check the Master's and the
application's<br/>spark.ui.holdEnabled, and the reported hold support
M->>C: SetApplicationHold (ask, not awaited, unchanged)
M-->>S: ApplicationHoldResponse(success, message)
S-->>R: HoldApplicationResponse (JSON)
C-->>M: ApplicationHoldUpdated (held/draining in /json/)
```
- `RequestApplicationHold` can now be asked as well as sent. When it is
asked, the Master replies with a new `ApplicationHoldResponse(success,
message)`. The Master web UI keeps `send`ing it, so the UI behaves as before.
Like `kill`, a Master that is not `ALIVE` rejects the request.
- The shared `Master.handleApplicationHold` now returns that response. It
rejects a request when either the Master or the application disables
`spark.ui.holdEnabled`, when the application has not reported that it can be
held, or when the application is unknown or finished. Otherwise it forwards the
request to the driver without waiting for the driver's answer, as before.
- Like `kill`, the response only tells whether the request was forwarded to
the driver, not its outcome. The driver drains its executors asynchronously,
and the outcome shows up in the `held` and `draining` fields of the Master's
`/json/` endpoint.
- No new configuration. The Master-wide `spark.ui.holdEnabled` gates both
the web UI controls and these actions.
- On the REST side, `HoldRequestServlet` and `StandaloneHoldRequestServlet`
relay the request to the Master the way the `kill` servlets do. A new
`HoldApplicationResponse` message carries `appId`, `success`, and `message`.
```bash
$ curl -XPOST http://IP:PORT/v1/submissions/hold/app-20260930120000-0000
{
"action" : "HoldApplicationResponse",
"appId" : "app-20260930120000-0000",
"message" : "Requesting application app-20260930120000-0000 to hold.",
"serverSparkVersion" : "4.4.0",
"success" : true
}
```
The docs add the two actions to the REST API table and describe them in
**Held Applications** next to the scripted web UI requests documented by
SPARK-59915:
- https://github.com/apache/spark/pull/59176
### Why are the changes needed?
Since SPARK-59353, the Master web UI's hold and resume endpoints require
the per-UI CSRF token, so a script has to read the token from the page before
sending the request:
- https://github.com/apache/spark/pull/58640
The REST Submission API is the Master's API for scripts and already offers
`kill`, `killall`, and `clear`. Holding is the graceful alternative to killing:
the application gives back its executors but keeps its driver and the shuffle
output already written. It belongs in the same API, without the token round
trip. These actions also reach client-mode applications, which the existing
REST `kill` cannot, because `kill` takes a submission ID.
### Does this PR introduce _any_ user-facing change?
Yes. The Standalone Master REST Submission API gets the new `hold` and
`resume` actions. Since hold and resume are new in the unreleased Apache Spark
4.4.0, there is no behavior change compared to the released versions.
Within the unreleased branches, the Master now rejects a hold request for
an application that has not reported that it can be held, instead of forwarding
it to the driver. The Master web UI never offers the controls for such an
application, so the UI is unaffected.
Like the other REST actions, `hold` and `resume` check no ACLs. The docs
recommend protecting the REST API with `JWSFilter` or disabling it with
`spark.master.rest.enabled=false`.
### How was this patch tested?
Pass the CIs with the newly added test cases:
- `StandaloneRestSubmitSuite`: `SPARK-59916: hold and resume an application`
- Both actions relay the Master's reply together with the application ID.
- A request without an application ID is rejected with 400.
- `AppClientSuite`: `SPARK-59916: hold and resume an application through an
ask to the Master`
- A real Master rejects the request, without reaching the driver, for an
application that has not reported that it can be held and for an unknown
application.
- Once the application reports that it can be held, the Master accepts
both requests and both reach the driver.
- `MasterSuite`: `SPARK-59916: Reject the hold and resume requests when the
Master disables holding`
The existing `SPARK-59061: hold and resume an application from the Master`
test now reports the hold support first, since the Master requires it. The JSON
in the documentation example is the actual output of `HoldApplicationResponse`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59178 from dongjoon-hyun/SPARK-59916.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../org/apache/spark/deploy/DeployMessage.scala | 4 ++
.../org/apache/spark/deploy/master/Master.scala | 55 +++++++++++++++++-----
.../spark/deploy/rest/RestSubmissionServer.scala | 35 +++++++++++++-
.../spark/deploy/rest/StandaloneRestServer.scala | 30 ++++++++++++
.../deploy/rest/SubmitRestProtocolResponse.scala | 12 +++++
.../org/apache/spark/internal/config/UI.scala | 3 +-
.../spark/deploy/client/AppClientSuite.scala | 49 ++++++++++++++++++-
.../apache/spark/deploy/master/MasterSuite.scala | 11 +++++
.../deploy/rest/StandaloneRestSubmitSuite.scala | 39 +++++++++++++++
docs/configuration.md | 4 +-
docs/spark-standalone.md | 36 ++++++++++++++
11 files changed, 260 insertions(+), 18 deletions(-)
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 28001f06d46f..1d6657b57f7e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
@@ -294,6 +294,10 @@ private[deploy] object DeployMessages {
def restUri: Option[String] = restPort.map { p => "spark://" + host + ":"
+ p }
}
+ // Reply to RequestApplicationHold when it is asked, as the REST server
does, telling whether
+ // the request was forwarded to the driver.
+ case class ApplicationHoldResponse(success: Boolean, message: String)
+
// WorkerWebUI to Worker
case object RequestWorkerState
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 1c3188b91407..58ae3fd8529e 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
@@ -34,6 +34,7 @@ import org.apache.spark.deploy.master.ui.MasterWebUI
import org.apache.spark.deploy.rest.StandaloneRestServer
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.MessageWithContext
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Deploy._
import org.apache.spark.internal.config.Deploy.WorkerSelectionPolicy._
@@ -129,6 +130,8 @@ private[deploy] class Master(
val historyServerUrl = conf.get(MASTER_UI_HISTORY_SERVER_URL)
val useAppNameAsAppId = conf.get(MASTER_USE_APP_NAME_AS_APP_ID)
val useDriverIdAsAppName = conf.get(MASTER_USE_DRIVER_ID_AS_APP_NAME)
+ // Whether applications can be held through this Master, from its web UI or
REST server
+ private val holdEnabled = conf.get(UI_HOLD_ENABLED)
// Alternative application submission gateway that is stable across Spark
versions
private val restServerEnabled = conf.get(MASTER_REST_SERVER_ENABLED)
@@ -534,6 +537,15 @@ private[deploy] class Master(
val formattedExecutorIds = formatExecutorIds(executorIds)
context.reply(handleKillExecutors(appId, formattedExecutorIds))
+ case RequestApplicationHold(appId, hold) =>
+ if (state != RecoveryState.ALIVE) {
+ val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
+ "Can only hold or resume applications in ALIVE state."
+ context.reply(ApplicationHoldResponse(success = false, msg))
+ } else {
+ context.reply(handleApplicationHold(appId, hold))
+ }
+
case DecommissionWorkersOnHosts(hostnames) =>
if (state != RecoveryState.STANDBY) {
context.reply(decommissionWorkersOnHosts(hostnames))
@@ -1235,25 +1247,41 @@ 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()`.
+ * Handle a hold or resume request made from the Master UI or the REST
server by forwarding it
+ * to the driver, which owns the hold: see `SparkContext.holdExecutors()`.
The request is
+ * rejected unless both this Master and the application enable
`spark.ui.holdEnabled`, and the
+ * application reported that it can be held.
*
- * 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.
+ * The driver asks this Master to lower its executor requirement 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. The
returned response,
+ * which the REST server relays, tells only whether the request was
forwarded.
*/
- private def handleApplicationHold(appId: String, hold: Boolean): Unit = {
+ private def handleApplicationHold(appId: String, hold: Boolean):
ApplicationHoldResponse = {
val action = if (hold) "hold" else "resume"
+ def reject(msg: MessageWithContext): ApplicationHoldResponse = {
+ logWarning(msg)
+ ApplicationHoldResponse(success = false, msg.message)
+ }
idToApp.get(appId) match {
+ case _ if !holdEnabled =>
+ // The Master UI offers no control then, but the REST server still
takes requests.
+ reject(log"Ignoring the ${MDC(LogKeys.OPERATION, action)} request for
application " +
+ log"${MDC(LogKeys.APP_ID, appId)}: ${MDC(LogKeys.CONFIG,
UI_HOLD_ENABLED.key)} is " +
+ log"disabled on the Master.")
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 " +
+ // The driver would reject it too; answer here without the round trip.
+ reject(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 && !app.holdSupported =>
+ // Act only once the driver reports that it can be held, as the UI
controls do.
+ reject(log"Ignoring the ${MDC(LogKeys.OPERATION, action)} request for
application " +
+ log"${MDC(LogKeys.APP_ID, appId)}, which has not reported that it
can be held.")
case Some(app) if !app.isFinished =>
- logInfo(log"Requesting application ${MDC(LogKeys.APP_ID, appId)} to " +
- log"${MDC(LogKeys.OPERATION, action)}.")
+ val msg = log"Requesting application ${MDC(LogKeys.APP_ID, appId)} to
" +
+ log"${MDC(LogKeys.OPERATION, action)}."
+ logInfo(msg)
app.driver.ask[Boolean](SetApplicationHold(hold)).onComplete {
case Success(acknowledged) =>
if (!acknowledged) {
@@ -1264,8 +1292,9 @@ private[deploy] class Master(
logWarning(log"Failed to ${MDC(LogKeys.OPERATION, action)}
application " +
log"${MDC(LogKeys.APP_ID, appId)}", t)
}(ThreadUtils.sameThread)
+ ApplicationHoldResponse(success = true, msg.message)
case _ =>
- logWarning(log"Ignoring the ${MDC(LogKeys.OPERATION, action)} request
for unknown or " +
+ reject(log"Ignoring the ${MDC(LogKeys.OPERATION, action)} request for
unknown or " +
log"finished application ${MDC(LogKeys.APP_ID, appId)}.")
}
}
diff --git
a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
index 2957be2c4372..88d98d0236a6 100644
---
a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
@@ -60,6 +60,8 @@ private[spark] abstract class RestSubmissionServer(
protected val submitRequestServlet: SubmitRequestServlet
protected val killRequestServlet: KillRequestServlet
protected val killAllRequestServlet: KillAllRequestServlet
+ protected val holdRequestServlet: HoldRequestServlet
+ protected val resumeRequestServlet: HoldRequestServlet
protected val statusRequestServlet: StatusRequestServlet
protected val clearRequestServlet: ClearRequestServlet
protected val readyzRequestServlet: ReadyzRequestServlet
@@ -73,6 +75,8 @@ private[spark] abstract class RestSubmissionServer(
s"$baseContext/create/*" -> submitRequestServlet,
s"$baseContext/kill/*" -> killRequestServlet,
s"$baseContext/killall/*" -> killAllRequestServlet,
+ s"$baseContext/hold/*" -> holdRequestServlet,
+ s"$baseContext/resume/*" -> resumeRequestServlet,
s"$baseContext/status/*" -> statusRequestServlet,
s"$baseContext/clear/*" -> clearRequestServlet,
s"$baseContext/readyz/*" -> readyzRequestServlet,
@@ -287,6 +291,33 @@ private[rest] abstract class KillAllRequestServlet extends
RestServlet {
protected def handleKillAll(): KillAllSubmissionResponse
}
+/**
+ * A servlet for handling hold or resume requests passed to the
[[RestSubmissionServer]].
+ *
+ * @param hold whether this servlet holds or resumes the application
+ */
+private[rest] abstract class HoldRequestServlet(hold: Boolean) extends
RestServlet {
+
+ /**
+ * If an application ID is specified in the URL, have the Master forward the
hold or resume
+ * request to the corresponding driver and return an appropriate response to
the client.
+ * Otherwise, return error.
+ */
+ protected override def doPost(
+ request: HttpServletRequest,
+ response: HttpServletResponse): Unit = {
+ val appId = parseSubmissionId(request.getPathInfo)
+ val responseMessage = appId.map(handleHold).getOrElse {
+ response.setStatus(HttpServletResponse.SC_BAD_REQUEST)
+ val action = if (hold) "hold" else "resume"
+ handleError(s"Application ID is missing in $action request.")
+ }
+ sendResponse(responseMessage, response)
+ }
+
+ protected def handleHold(appId: String): HoldApplicationResponse
+}
+
/**
* A servlet for handling clear requests passed to the
[[RestSubmissionServer]].
*/
@@ -439,8 +470,8 @@ private class ErrorServlet extends RestServlet {
"Missing the /submissions prefix."
case `serverVersion` :: "submissions" :: tail =>
// http://host:port/correct-version/submissions/*
- "Missing an action: please specify one of /create, /kill, /killall,
/clear, /status, " +
- "or /readyz."
+ "Missing an action: please specify one of /create, /kill, /killall,
/hold, /resume, " +
+ "/clear, /status, or /readyz."
case unknownVersion :: tail =>
// http://host:port/unknown-version/*
versionMismatch = true
diff --git
a/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala
b/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala
index 79a54bae4048..14c6704ca3d7 100644
---
a/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/rest/StandaloneRestServer.scala
@@ -67,6 +67,10 @@ private[deploy] class StandaloneRestServer(
new StandaloneKillRequestServlet(masterEndpoint, masterConf)
protected override val killAllRequestServlet =
new StandaloneKillAllRequestServlet(masterEndpoint, masterConf)
+ protected override val holdRequestServlet =
+ new StandaloneHoldRequestServlet(masterEndpoint, masterConf, hold = true)
+ protected override val resumeRequestServlet =
+ new StandaloneHoldRequestServlet(masterEndpoint, masterConf, hold = false)
protected override val statusRequestServlet =
new StandaloneStatusRequestServlet(masterEndpoint, masterConf)
protected override val clearRequestServlet =
@@ -110,6 +114,32 @@ private[rest] class
StandaloneKillAllRequestServlet(masterEndpoint: RpcEndpointR
}
}
+/**
+ * A servlet for handling hold or resume requests passed to the
[[StandaloneRestServer]].
+ *
+ * The Master checks `spark.ui.holdEnabled` on both itself and the
application, like for the
+ * Master UI controls. The response tells only whether the request was
forwarded to the driver:
+ * the resulting hold status is reported by the Master's `/json/` endpoint
once the driver acts
+ * on it.
+ */
+private[rest] class StandaloneHoldRequestServlet(
+ masterEndpoint: RpcEndpointRef,
+ conf: SparkConf,
+ hold: Boolean)
+ extends HoldRequestServlet(hold) {
+
+ protected def handleHold(appId: String): HoldApplicationResponse = {
+ val response =
masterEndpoint.askSync[DeployMessages.ApplicationHoldResponse](
+ DeployMessages.RequestApplicationHold(appId, hold))
+ val h = new HoldApplicationResponse
+ h.serverSparkVersion = sparkVersion
+ h.message = response.message
+ h.appId = appId
+ h.success = response.success
+ h
+ }
+}
+
/**
* A servlet for handling status requests passed to the
[[StandaloneRestServer]].
*/
diff --git
a/core/src/main/scala/org/apache/spark/deploy/rest/SubmitRestProtocolResponse.scala
b/core/src/main/scala/org/apache/spark/deploy/rest/SubmitRestProtocolResponse.scala
index 4c484419d3be..239e800ea8af 100644
---
a/core/src/main/scala/org/apache/spark/deploy/rest/SubmitRestProtocolResponse.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/rest/SubmitRestProtocolResponse.scala
@@ -65,6 +65,18 @@ private[spark] class KillAllSubmissionResponse extends
SubmitRestProtocolRespons
}
}
+/**
+ * A response to a hold or resume request in the REST application submission
protocol.
+ */
+private[spark] class HoldApplicationResponse extends
SubmitRestProtocolResponse {
+ var appId: String = null
+ protected override def doValidate(): Unit = {
+ super.doValidate()
+ assertFieldIsSet(appId, "appId")
+ assertFieldIsSet(success, "success")
+ }
+}
+
/**
* A response to a clear request in the REST application submission protocol.
*/
diff --git a/core/src/main/scala/org/apache/spark/internal/config/UI.scala
b/core/src/main/scala/org/apache/spark/internal/config/UI.scala
index e5596d0c6bfc..4c2e328bf0c0 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/UI.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/UI.scala
@@ -99,7 +99,8 @@ private[spark] object UI {
"spark.decommission.enabled is true, the shuffle data is kept outside
the executors " +
"(through either spark.shuffle.service.enabled or a ShuffleDataIO with
reliable " +
"storage), and the cluster manager can hold executors: Standalone, YARN,
and " +
- "Kubernetes with spark.kubernetes.allocation.pods.allocator=direct.")
+ "Kubernetes with spark.kubernetes.allocation.pods.allocator=direct. On a
Standalone " +
+ "Master, this also gates the hold and resume actions of the Master REST
API.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.booleanConf
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 814d68587a83..6d8e8b0944d6 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
@@ -28,7 +28,7 @@ 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,
RequestApplicationHold, RequestMasterState, WorkerDecommissioning}
+import org.apache.spark.deploy.DeployMessages.{ApplicationHoldResponse,
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}
@@ -264,6 +264,10 @@ class AppClientSuite
assert(getApplications().length === 1, "master should have 1
registered app")
}
val appId = getApplications().head.id
+ ci.client.reportHoldStatus(supported = true, held = false)
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().head.holdSupported)
+ }
// A hold requested from the Master UI reaches the driver.
master.self.send(RequestApplicationHold(appId, hold = true))
@@ -310,6 +314,49 @@ class AppClientSuite
}
}
+ test("SPARK-59916: hold and resume an application through an ask 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")
+ }
+ val appId = getApplications().head.id
+
+ // The REST server asks rather than sends, and the reply tells whether
the request was
+ // forwarded to the driver.
+ def askHold(id: String, hold: Boolean): ApplicationHoldResponse =
+
master.self.askSync[ApplicationHoldResponse](RequestApplicationHold(id, hold))
+
+ // Rejected until the driver reports that it can be held.
+ val unsupported = askHold(appId, hold = true)
+ assert(!unsupported.success)
+ assert(unsupported.message.contains("has not reported that it can be
held"))
+
+ val unknown = askHold("unknown-app", hold = true)
+ assert(!unknown.success)
+ assert(unknown.message.contains("unknown or finished application"))
+ assert(ci.listener.holdRequestList.isEmpty)
+
+ ci.client.reportHoldStatus(supported = true, held = false)
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(getApplications().head.holdSupported)
+ }
+ Seq(true, false).foreach { hold =>
+ assert(askHold(appId, hold).success)
+ }
+ 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("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 2b68f3db1b44..4325c83444ec 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
@@ -28,6 +28,7 @@ import org.apache.spark.deploy._
import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Deploy._
+import org.apache.spark.internal.config.UI.UI_HOLD_ENABLED
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.rpc.{RpcAddress, RpcEndpoint, RpcEnv}
@@ -284,6 +285,16 @@ class MasterSuite extends MasterSuiteBase {
assert(appInfo.numDrainingExecutors === 0)
}
+ test("SPARK-59916: Reject the hold and resume requests when the Master
disables holding") {
+ val master = makeAliveMaster(new SparkConf().set(UI_HOLD_ENABLED, false))
+ Seq(true, false).foreach { hold =>
+ val response = master.self.askSync[ApplicationHoldResponse](
+ RequestApplicationHold("app-1", hold))
+ assert(!response.success)
+ assert(response.message.contains(s"${UI_HOLD_ENABLED.key} is disabled on
the Master"))
+ }
+ }
+
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/rest/StandaloneRestSubmitSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala
index 39ec9fc5aa7e..5ab6503afadb 100644
---
a/core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala
+++
b/core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala
@@ -267,6 +267,30 @@ class StandaloneRestSubmitSuite extends SparkFunSuite {
assert(errorResponse.message.contains("Master is not ready."))
}
+ test("SPARK-59916: hold and resume an application") {
+ val masterUrl = startDummyServer()
+ val httpUrl = masterUrl.replace("spark://", "http://";)
+ val v = RestSubmissionServer.PROTOCOL_VERSION
+ Seq("hold" -> "held", "resume" -> "resumed").foreach { case (action,
message) =>
+ val path = s"$httpUrl/$v/submissions/$action"
+ val (response, code) = sendHttpRequestWithResponse(s"$path/my-app",
"POST")
+ assert(code === HttpServletResponse.SC_OK)
+ val holdResponse = getHoldResponse(response)
+ assert(holdResponse.action === Utils.getFormattedClassName(holdResponse))
+ assert(holdResponse.serverSparkVersion === SPARK_VERSION)
+ assert(holdResponse.message === message)
+ assert(holdResponse.appId === "my-app")
+ assert(holdResponse.success)
+
+ // A request without an application ID is rejected.
+ Seq(path, s"$path/").foreach { badPath =>
+ val (badResponse, badCode) = sendHttpRequestWithResponse(badPath,
"POST")
+ assert(badCode === HttpServletResponse.SC_BAD_REQUEST)
+ getErrorResponse(badResponse)
+ }
+ }
+ }
+
/* ---------------------------------------- *
| Aberrant client / server behavior |
* ---------------------------------------- */
@@ -780,6 +804,15 @@ class StandaloneRestSubmitSuite extends SparkFunSuite {
}
}
+ /** Return the response as a hold response, or fail with error otherwise. */
+ private def getHoldResponse(response: SubmitRestProtocolResponse):
HoldApplicationResponse = {
+ response match {
+ case h: HoldApplicationResponse => h
+ case e: ErrorResponse => fail(s"Server returned error: ${e.message}")
+ case r => fail(s"Expected hold response. Actual: ${r.toJson}")
+ }
+ }
+
/** Return the response as a clear response, or fail with error otherwise. */
private def getClearResponse(response: SubmitRestProtocolResponse):
ClearResponse = {
response match {
@@ -874,6 +907,8 @@ private class DummyMaster(
context.reply(KillDriverResponse(self, driverId, success = true,
killMessage))
case RequestKillAllDrivers =>
context.reply(KillAllDriversResponse(self, success = true, killMessage))
+ case RequestApplicationHold(_, hold) =>
+ context.reply(ApplicationHoldResponse(success = true, if (hold) "held"
else "resumed"))
case RequestDriverStatus(driverId) =>
context.reply(DriverStatusResponse(found = true, Some(state), None,
None, exception))
case RequestClearCompletedDriversAndApps =>
@@ -939,6 +974,10 @@ private class FaultyStandaloneRestServer(
protected override val submitRequestServlet = new MalformedSubmitServlet
protected override val killRequestServlet = new InvalidKillServlet
protected override val killAllRequestServlet = new InvalidKillAllServlet
+ protected override val holdRequestServlet =
+ new StandaloneHoldRequestServlet(masterEndpoint, masterConf, hold = true)
+ protected override val resumeRequestServlet =
+ new StandaloneHoldRequestServlet(masterEndpoint, masterConf, hold = false)
protected override val statusRequestServlet = new ExplodingStatusServlet
protected override val clearRequestServlet = new ExplodingClearServlet
protected override val readyzRequestServlet = new ExplodingReadyzServlet
diff --git a/docs/configuration.md b/docs/configuration.md
index fef5d4b3ba2a..c85e1f58f1aa 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1750,7 +1750,9 @@ Apart from these, the following properties are also
available, and may be useful
executors (through either <code>spark.shuffle.service.enabled</code> or a
<code>ShuffleDataIO</code> with reliable storage), and the cluster manager
can hold
executors: Standalone, YARN, and Kubernetes with
- <code>spark.kubernetes.allocation.pods.allocator=direct</code>.
+ <code>spark.kubernetes.allocation.pods.allocator=direct</code>. On a
Standalone Master,
+ this also gates the <code>hold</code> and <code>resume</code> actions of
the
+ <a href="spark-standalone.html#rest-api">Master REST API</a>.
</td>
<td>4.4.0</td>
</tr>
diff --git a/docs/spark-standalone.md b/docs/spark-standalone.md
index 1024f0f67052..2e588a8f1657 100644
--- a/docs/spark-standalone.md
+++ b/docs/spark-standalone.md
@@ -653,6 +653,20 @@ via
<code>http://[host:port]/[version]/submissions/[action]</code> where
<td>Kill all running Spark drivers.</td>
<td>4.0.0</td>
</tr>
+ <tr>
+ <td><code>hold</code></td>
+ <td>POST</td>
+ <td>Hold a running Spark application, given its application ID rather than
a submission ID.
+ See <a href="#held-applications">Held Applications</a>.</td>
+ <td>4.4.0</td>
+ </tr>
+ <tr>
+ <td><code>resume</code></td>
+ <td>POST</td>
+ <td>Resume a held Spark application, given its application ID rather than
a submission ID.
+ See <a href="#held-applications">Held Applications</a>.</td>
+ <td>4.4.0</td>
+ </tr>
<tr>
<td><code>status</code></td>
<td>GET</td>
@@ -819,6 +833,28 @@ If the Master web UI requires authentication, reading the
token and sending the
need the same credentials. The request is forwarded to the driver
asynchronously, so its outcome
shows up afterwards in the `held` and `draining` fields of the Master's
`/json/` endpoint.
+Scripts can also use the `hold` and `resume` actions of the [REST
API](#rest-api), which need no
+CSRF token. They take the application ID rather than a submission ID, so they
work for
+applications submitted in client mode too, and they are gated the same way, by
+`spark.ui.holdEnabled` on the Master and on the application. The response only
tells whether the
+request was forwarded to the driver; poll the `held` and `draining` fields of
the Master's
+`/json/` endpoint for the outcome.
+
+```bash
+$ curl -XPOST http://IP:PORT/v1/submissions/hold/app-20260930120000-0000
+{
+ "action" : "HoldApplicationResponse",
+ "appId" : "app-20260930120000-0000",
+ "message" : "Requesting application app-20260930120000-0000 to hold.",
+ "serverSparkVersion" : "4.4.0",
+ "success" : true
+}
+```
+
+Like the other actions of the REST API, `hold` and `resume` check no ACLs, and
a browser can be
+made to send them from another site. Require a signed `Authorization` header
with `JWSFilter` as
+described in [REST API](#rest-api), or disable the REST API with
`spark.master.rest.enabled=false`.
+
# Running Alongside Hadoop
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
