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 f2346e1327ca [SPARK-58049][CORE] Limit the request body size of the
Master REST submission server
f2346e1327ca is described below
commit f2346e1327cadc8d5e8208f87aa9886699488532
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Jul 9 08:00:43 2026 -0700
[SPARK-58049][CORE] Limit the request body size of the Master REST
submission server
### What changes were proposed in this pull request?
`SubmitRequestServlet.doPost` in the Master REST submission server used to
read the
entire request body into a `String` before any validation. This PR bounds
that read
by a new config `spark.master.rest.maxRequestBodySize` (default `100m`): it
rejects
early when `Content-Length` exceeds the limit, and otherwise reads at most
`limit + 1`
bytes so an oversized body is caught even when `Content-Length` is missing
or wrong.
Over-limit requests get HTTP `413` (`SC_REQUEST_ENTITY_TOO_LARGE`).
### Why are the changes needed?
To improve the performance and robustness of Spark Master REST API.
- **Performance**: Reading raw bytes in bulk 8 KiB chunks and decoding once
via `new String(bytes, UTF_8)` is faster than `scala.io.Source`, which iterates
and decodes character-by-character through a `StringBuilder`. It also decodes
as UTF-8 explicitly rather than relying on the platform default charset.
- **Memory**: The body read is now bounded instead of unbounded. Previously
`Source.fromInputStream(...).mkString` buffered the entire request body
regardless of size, so a multi-GB request could OOM the Master. Now at most
limit + 1 bytes (default 100 MiB) are read before an over-limit request is
rejected and discarded. Note that `readNBytes(len)` does not pre-allocate `len`
bytes — it accumulates 8 KiB chunks proportional to the bytes actually read —
so passing Int.MaxValue as the c [...]
### Does this PR introduce _any_ user-facing change?
- The default is generous enough that legitimate submissions are unaffected.
- However, yes, technically. A submit request whose body exceeds
`spark.master.rest.maxRequestBodySize`
(default 100 MiB) is now rejected with HTTP 413. So, the migration guide is
updated.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #57138 from dongjoon-hyun/SPARK-58049.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/deploy/rest/RestSubmissionServer.scala | 33 +++++++++--
.../spark/deploy/rest/StandaloneRestServer.scala | 4 ++
.../org/apache/spark/internal/config/package.scala | 11 ++++
.../deploy/rest/StandaloneRestSubmitSuite.scala | 69 ++++++++++++++++++++--
docs/core-migration-guide.md | 2 +
5 files changed, 111 insertions(+), 8 deletions(-)
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 ee404ac11a96..2957be2c4372 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
@@ -17,11 +17,10 @@
package org.apache.spark.deploy.rest
+import java.nio.charset.StandardCharsets
import java.util.EnumSet
import java.util.concurrent.{Executors, ExecutorService}
-import scala.io.Source
-
import com.fasterxml.jackson.core.JsonProcessingException
import jakarta.servlet.DispatcherType
import jakarta.servlet.http.{HttpServlet, HttpServletRequest,
HttpServletResponse}
@@ -34,7 +33,7 @@ import org.json4s.jackson.JsonMethods._
import org.apache.spark.{SPARK_VERSION => sparkVersion, SparkConf}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.internal.config.{MASTER_REST_SERVER_FILTERS,
MASTER_REST_SERVER_MAX_THREADS, MASTER_REST_SERVER_VIRTUAL_THREADS}
+import org.apache.spark.internal.config.{MASTER_REST_SERVER_FILTERS,
MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE, MASTER_REST_SERVER_MAX_THREADS,
MASTER_REST_SERVER_VIRTUAL_THREADS}
import org.apache.spark.util.Utils
/**
@@ -43,6 +42,7 @@ import org.apache.spark.util.Utils
* This server responds with different HTTP codes depending on the situation:
* 200 OK - Request was processed successfully
* 400 BAD REQUEST - Request was malformed, not successfully validated, or
of unexpected type
+ * 413 REQUEST ENTITY TOO LARGE - Request body exceeded the configured
maximum size
* 468 UNKNOWN PROTOCOL VERSION - Request specified a protocol this server
does not understand
* 500 INTERNAL SERVER ERROR - Server throws an exception internally while
processing the request
*
@@ -358,6 +358,24 @@ private[rest] abstract class StatusRequestServlet extends
RestServlet {
*/
private[rest] abstract class SubmitRequestServlet extends RestServlet {
+ /** The maximum size in bytes of a request body this servlet will read. */
+ protected def maxRequestSizeBytes: Long =
+ MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE.defaultValue.get
+
+ /**
+ * Read the request body as a UTF-8 string, or return `None` if it exceeds
`limit` bytes.
+ * The read is bounded so an oversized body cannot exhaust memory even when
the
+ * Content-Length header is missing or understates the actual size.
+ */
+ private def readRequestBody(request: HttpServletRequest, limit: Long):
Option[String] = {
+ if (request.getContentLengthLong > limit) {
+ None
+ } else {
+ val bytes = request.getInputStream.readNBytes(math.min(limit + 1,
Int.MaxValue).toInt)
+ if (bytes.length > limit) None else Some(new String(bytes,
StandardCharsets.UTF_8))
+ }
+ }
+
/**
* Submit an application to the Master with parameters specified in the
request.
*
@@ -368,9 +386,16 @@ private[rest] abstract class SubmitRequestServlet extends
RestServlet {
protected override def doPost(
requestServlet: HttpServletRequest,
responseServlet: HttpServletResponse): Unit = {
+ val limit = maxRequestSizeBytes
+ val requestMessageJson = readRequestBody(requestServlet, limit).getOrElse {
+
responseServlet.setStatus(HttpServletResponse.SC_REQUEST_ENTITY_TOO_LARGE)
+ val msg = s"The request body exceeds the maximum allowed size ($limit
bytes). " +
+ s"Adjust ${MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE.key} to change the
limit."
+ sendResponse(handleError(msg), responseServlet)
+ return
+ }
val responseMessage =
try {
- val requestMessageJson =
Source.fromInputStream(requestServlet.getInputStream).mkString
val requestMessage =
SubmitRestProtocolMessage.fromJson(requestMessageJson)
// The response should have already been validated on the client.
// In case this is not true, validate it ourselves to avoid potential
NPEs.
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 bb598b042958..400eef2fd3e5 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
@@ -174,6 +174,10 @@ private[rest] class StandaloneSubmitRequestServlet(
conf: SparkConf)
extends SubmitRequestServlet {
+ override protected def maxRequestSizeBytes: Long =
+ Option(conf).map(_.get(config.MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE))
+ .getOrElse(super.maxRequestSizeBytes)
+
private def replacePlaceHolder(variable: String) = variable match {
case s"{{$name}}" if System.getenv(name) != null => System.getenv(name)
case _ => variable
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala
b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 79ac503e69be..b658c31753e3 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -2106,6 +2106,17 @@ package object config {
.toSequence
.createWithDefault(Nil)
+ private[spark] val MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE =
+ ConfigBuilder("spark.master.rest.maxRequestBodySize")
+ .doc("The maximum size of the request body accepted by the Spark Master
REST API. " +
+ "Requests whose body exceeds this size are rejected with HTTP 413 " +
+ "(Request Entity Too Large).")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .bytesConf(ByteUnit.BYTE)
+ .checkValue(_ > 0, "The max request body size must be positive.")
+ .createWithDefaultString("100m")
+
private[spark] val MASTER_UI_PORT = ConfigBuilder("spark.master.ui.port")
.version("1.1.0")
.intConf
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 bf4a7c4f4d9d..a912926e64d5 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
@@ -35,7 +35,7 @@ import org.apache.spark.deploy.{SparkSubmit,
SparkSubmitArguments}
import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.master.DriverState._
import org.apache.spark.deploy.master.RecoveryState
-import
org.apache.spark.internal.config.{MASTER_REST_SERVER_ALLOWED_APP_RESOURCE_PATTERNS,
MASTER_REST_SERVER_FILTERS, MASTER_REST_SERVER_MAX_THREADS,
MASTER_REST_SERVER_VIRTUAL_THREADS}
+import
org.apache.spark.internal.config.{MASTER_REST_SERVER_ALLOWED_APP_RESOURCE_PATTERNS,
MASTER_REST_SERVER_FILTERS, MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE,
MASTER_REST_SERVER_MAX_THREADS, MASTER_REST_SERVER_VIRTUAL_THREADS}
import org.apache.spark.rpc._
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils
@@ -599,6 +599,62 @@ class StandaloneRestSubmitSuite extends SparkFunSuite {
}
}
+ test("SPARK-58049: Reject an over-sized request body with
SC_REQUEST_ENTITY_TOO_LARGE") {
+ val conf = new SparkConf()
+ conf.set(MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE.key, "1k")
+ val localhost = Utils.localHostName()
+ val securityManager = new SecurityManager(conf)
+ rpcEnv =
+ Some(RpcEnv.create("rest-with-maxRequestBodySize", localhost, 0, conf,
securityManager))
+ val fakeMasterRef = rpcEnv.get.setupEndpoint("fake-master", new
DummyMaster(rpcEnv.get))
+ val _server = new StandaloneRestServer(localhost, 0, conf, fakeMasterRef,
"spark://fake:7077")
+ server = Some(_server)
+ val port = _server.start()
+ val v = RestSubmissionServer.PROTOCOL_VERSION
+ val path = s"http://$localhost:$port/$v/submissions/create";
+ val (response, code) = sendHttpRequestWithResponse(path, "POST", "x" *
4096)
+ assert(code === HttpServletResponse.SC_REQUEST_ENTITY_TOO_LARGE)
+ getErrorResponse(response)
+ }
+
+ test("SPARK-58049: Reject an over-sized request body without
Content-Length") {
+ val conf = new SparkConf()
+ conf.set(MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE.key, "1k")
+ val localhost = Utils.localHostName()
+ val securityManager = new SecurityManager(conf)
+ rpcEnv =
+ Some(RpcEnv.create("rest-with-maxRequestBodySize", localhost, 0, conf,
securityManager))
+ val fakeMasterRef = rpcEnv.get.setupEndpoint("fake-master", new
DummyMaster(rpcEnv.get))
+ val _server = new StandaloneRestServer(localhost, 0, conf, fakeMasterRef,
"spark://fake:7077")
+ server = Some(_server)
+ val port = _server.start()
+ val v = RestSubmissionServer.PROTOCOL_VERSION
+ val path = s"http://$localhost:$port/$v/submissions/create";
+ val (response, code) = sendHttpRequestWithResponse(path, "POST", "x" *
4096, chunked = true)
+ assert(code === HttpServletResponse.SC_REQUEST_ENTITY_TOO_LARGE)
+ getErrorResponse(response)
+ }
+
+ test("SPARK-58049: Accept a request body exactly at the limit") {
+ val json = constructSubmitRequest("spark://localhost:7077").toJson
+ val limit = json.getBytes(StandardCharsets.UTF_8).length
+ val conf = new SparkConf()
+ conf.set(MASTER_REST_SERVER_MAX_REQUEST_BODY_SIZE.key, limit.toString)
+ val localhost = Utils.localHostName()
+ val securityManager = new SecurityManager(conf)
+ rpcEnv =
+ Some(RpcEnv.create("rest-with-maxRequestBodySize", localhost, 0, conf,
securityManager))
+ val fakeMasterRef = rpcEnv.get.setupEndpoint("fake-master", new
DummyMaster(rpcEnv.get))
+ val _server = new StandaloneRestServer(localhost, 0, conf, fakeMasterRef,
"spark://fake:7077")
+ server = Some(_server)
+ val port = _server.start()
+ val v = RestSubmissionServer.PROTOCOL_VERSION
+ val path = s"http://$localhost:$port/$v/submissions/create";
+ val (response, code) = sendHttpRequestWithResponse(path, "POST", json)
+ assert(code === HttpServletResponse.SC_OK)
+ getSubmitResponse(response)
+ }
+
/* --------------------- *
| Helper methods |
* --------------------- */
@@ -740,11 +796,15 @@ class StandaloneRestSubmitSuite extends SparkFunSuite {
private def sendHttpRequest(
url: String,
method: String,
- body: String = ""): HttpURLConnection = {
+ body: String = "",
+ chunked: Boolean = false): HttpURLConnection = {
val conn = new
URI(url).toURL.openConnection().asInstanceOf[HttpURLConnection]
conn.setRequestMethod(method)
if (body.nonEmpty) {
conn.setDoOutput(true)
+ if (chunked) {
+ conn.setChunkedStreamingMode(0)
+ }
val out = new DataOutputStream(conn.getOutputStream)
out.write(body.getBytes(StandardCharsets.UTF_8))
out.close()
@@ -759,8 +819,9 @@ class StandaloneRestSubmitSuite extends SparkFunSuite {
private def sendHttpRequestWithResponse(
url: String,
method: String,
- body: String = ""): (SubmitRestProtocolResponse, Int) = {
- val conn = sendHttpRequest(url, method, body)
+ body: String = "",
+ chunked: Boolean = false): (SubmitRestProtocolResponse, Int) = {
+ val conn = sendHttpRequest(url, method, body, chunked)
(new RestSubmissionClient("spark://host:port").readResponse(conn),
conn.getResponseCode)
}
}
diff --git a/docs/core-migration-guide.md b/docs/core-migration-guide.md
index e80d04238ca0..f91abaa9ddca 100644
--- a/docs/core-migration-guide.md
+++ b/docs/core-migration-guide.md
@@ -36,6 +36,8 @@ license: |
- Since Spark 4.3, `spark.ui.allowFramingFrom` now uses CSP `frame-ancestors`
instead of the deprecated `X-Frame-Options: ALLOW-FROM` (which was ignored by
all modern browsers). This setting only takes effect when
`spark.ui.contentSecurityPolicy.enabled=true` (the default). When CSP is
disabled, `X-Frame-Options: SAMEORIGIN` is always used regardless of the
`allowFramingFrom` value.
+- Since Spark 4.3, the Spark Master REST API rejects a submission whose
request body exceeds `spark.master.rest.maxRequestBodySize` (default `100m`)
with HTTP 413. To allow larger request bodies, increase
`spark.master.rest.maxRequestBodySize`.
+
## Upgrading from Core 4.1 to 4.2
- Since Spark 4.2, Spark Master REST API uses Java 21 virtual threads by
default when running on Java 21 or later. To restore the legacy behavior, you
can set `spark.master.rest.virtualThread.enabled` to `false`.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
