Skip to content

[SPARK-50381][CORE] Support spark.master.rest.maxThreads #48921

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dongjoon-hyun wants to merge 1 commit into from
Closed

[SPARK-50381][CORE] Support spark.master.rest.maxThreads #48921

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import org.json4s.jackson.JsonMethods._
import org.apache.spark.{SPARK_VERSION => sparkVersion, SparkConf}
import org.apache.spark.internal.{Logging, MDC}
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.MASTER_REST_SERVER_FILTERS
import org.apache.spark.internal.config.{MASTER_REST_SERVER_FILTERS, MASTER_REST_SERVER_MAX_THREADS}
import org.apache.spark.util.Utils

/**
Expand Down Expand Up @@ -63,7 +63,8 @@ private[spark] abstract class RestSubmissionServer(
protected val clearRequestServlet: ClearRequestServlet
protected val readyzRequestServlet: ReadyzRequestServlet

private var _server: Option[Server] = None
// Visible for testing
private[rest] var _server: Option[Server] = None

// A mapping from URL prefixes to servlets that serve them. Exposed for testing.
protected val baseContext = s"/${RestSubmissionServer.PROTOCOL_VERSION}/submissions"
Expand Down Expand Up @@ -91,7 +92,7 @@ private[spark] abstract class RestSubmissionServer(
* Return a 2-tuple of the started server and the bound port.
*/
private def doStart(startPort: Int): (Server, Int) = {
val threadPool = new QueuedThreadPool
val threadPool = new QueuedThreadPool(masterConf.get(MASTER_REST_SERVER_MAX_THREADS))
threadPool.setDaemon(true)
val server = new Server(threadPool)

Expand Down
6 changes: 6 additions & 0 deletions core/src/main/scala/org/apache/spark/internal/config/package.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1987,6 +1987,12 @@ package object config {
.intConf
.createWithDefault(6066)

private[spark] val MASTER_REST_SERVER_MAX_THREADS = ConfigBuilder("spark.master.rest.maxThreads")
.doc("Maximum number of threads to use in the Spark Master REST API Server.")
.version("4.0.0")
.intConf
.createWithDefault(200)
Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 21, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default value is identical with the previously-used Jetty's default value.


private[spark] val MASTER_REST_SERVER_FILTERS = ConfigBuilder("spark.master.rest.filters")
.doc("Comma separated list of filter class names to apply to the Spark Master REST API.")
.version("4.0.0")
Expand Down
16 changes: 15 additions & 1 deletion core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import java.util.Base64
import scala.collection.mutable

import jakarta.servlet.http.HttpServletResponse
import org.eclipse.jetty.util.thread.ThreadPool.SizedThreadPool
import org.json4s.JsonAST._
import org.json4s.jackson.JsonMethods._

Expand All @@ -33,7 +34,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_FILTERS
import org.apache.spark.internal.config.{MASTER_REST_SERVER_FILTERS, MASTER_REST_SERVER_MAX_THREADS}
import org.apache.spark.rpc._
import org.apache.spark.util.ArrayImplicits._
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -545,6 +546,19 @@ class StandaloneRestSubmitSuite extends SparkFunSuite {
}
}

test("SPARK-50381: Support spark.master.rest.maxThreads") {
val conf = new SparkConf()
val localhost = Utils.localHostName()
val securityManager = new SecurityManager(conf)
rpcEnv = Some(RpcEnv.create("rest-with-maxThreads", localhost, 0, conf, securityManager))
val fakeMasterRef = rpcEnv.get.setupEndpoint("fake-master", new DummyMaster(rpcEnv.get))
conf.set(MASTER_REST_SERVER_MAX_THREADS, 2000)
server = Some(new StandaloneRestServer(localhost, 0, conf, fakeMasterRef, "spark://fake:7077"))
server.get.start()
val pool = server.get._server.get.getThreadPool.asInstanceOf[SizedThreadPool]
assert(pool.getMaxThreads === 2000)
}

/* --------------------- *
| Helper methods |
* --------------------- */
Expand Down