This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new fcc4a28d5844 [SPARK-58886][CORE] Fix Int overflow in
`CoarseGrainedSchedulerBackend.requestExecutors`
fcc4a28d5844 is described below
commit fcc4a28d58441891cb2f02a43ab184caa7b23c2d
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Aug 19 16:42:47 2026 -0700
[SPARK-58886][CORE] Fix Int overflow in
`CoarseGrainedSchedulerBackend.requestExecutors`
This PR fixes an `Int` overflow in
`CoarseGrainedSchedulerBackend.requestExecutors`:
the new total is now computed with `Long` arithmetic and saturates at
`Int.MaxValue`
instead of overflowing to a negative value. The executor request time
bookkeeping
(`updateExecRequestTime`) is also given the actually-applied delta, so a
saturated
request does not record an entry for an increase that never happened.
A test-only accessor `getRequestedTotalExecutors()` is added so the test
can observe
the requested totals.
The defect is reproducible on `master` when a user requests a huge total
first, e.g.
`sc.requestTotalExecutors(Int.MaxValue, 0, Map.empty)`. A subsequent
`requestExecutors(n)` overflows `numExisting + numAdditionalExecutors` to a
negative
number, which is then sent to the cluster manager unvalidated — the
negative-value
check in `requestTotalExecutors` does not cover this path.
No. It only prevents a pathological bookkeeping value in the corner case
above.
A new test in `CoarseGrainedSchedulerBackendSuite` using a
`local-cluster[0, 3, 1024]`
backend: `requestExecutors(1)` after `requestTotalExecutors(Int.MaxValue)`
saturates
the recorded total at `Int.MaxValue` instead of going negative. The test
fails without
the fix.
Generated-by: Claude Fable 5
Closes #58137 from dongjoon-hyun/SPARK-58886.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 77a4f757790c113405a0314bc870f2ea65da2391)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../cluster/CoarseGrainedSchedulerBackend.scala | 13 ++++++--
.../CoarseGrainedSchedulerBackendSuite.scala | 39 ++++++++++++++++++++++
2 files changed, 50 insertions(+), 2 deletions(-)
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 49590f0940f7..34c24eb79bb5 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
@@ -798,6 +798,11 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
execDataOption.map(_.resourceProfileId).getOrElse(ResourceProfile.UNKNOWN_RESOURCE_PROFILE_ID)
}
+ // this function is for testing only
+ private[spark] def getRequestedTotalExecutors(): Map[ResourceProfile, Int] =
synchronized {
+ requestedTotalExecutorsPerResourceProfile.toMap
+ }
+
/**
* Request an additional number of executors from the cluster manager. This
is
* requesting against the default ResourceProfile, we will need an API
change to
@@ -816,9 +821,13 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
val response = synchronized {
val defaultProf =
scheduler.sc.resourceProfileManager.defaultResourceProfile
val numExisting =
requestedTotalExecutorsPerResourceProfile.getOrElse(defaultProf, 0)
- requestedTotalExecutorsPerResourceProfile(defaultProf) = numExisting +
numAdditionalExecutors
+ // Saturate instead of overflowing when the current requirement is
already huge (e.g.
+ // Int.MaxValue after an unbounded `requestTotalExecutors`).
+ val newTotal =
+ math.min(numExisting.toLong + numAdditionalExecutors,
Int.MaxValue.toLong).toInt
+ requestedTotalExecutorsPerResourceProfile(defaultProf) = newTotal
// Account for executors pending to be added or removed
- updateExecRequestTime(defaultProf.id, numAdditionalExecutors)
+ updateExecRequestTime(defaultProf.id, newTotal - numExisting)
doRequestTotalExecutors(requestedTotalExecutorsPerResourceProfile.toMap)
}
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
index 127e4b2e413c..75c97b79cc09 100644
---
a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++
b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -606,6 +606,45 @@ class CoarseGrainedSchedulerBackendSuite extends
SparkFunSuite with LocalSparkCo
assert(mockEndpointRef.decommissionReceived)
}
+ test("SPARK-58886: requestExecutors should saturate instead of overflowing a
huge" +
+ " requested total") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+
+ sc = new SparkContext(conf)
+ val backend =
sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+
+ sc.requestTotalExecutors(Int.MaxValue - 1, 0, Map.empty)
+ backend.requestExecutors(2)
+
+ val defaultProf = sc.resourceProfileManager.defaultResourceProfile
+ assert(backend.getRequestedTotalExecutors()(defaultProf) === Int.MaxValue)
+
+ // Only the applied increase (1, not the requested 2) may be recorded as a
pending
+ // request time. Shrink the total to 1 to consume the huge seed entry,
leaving just
+ // that increment: exactly one of the two executors registered below
should get a
+ // request time.
+ sc.requestTotalExecutors(1, 0, Map.empty)
+
+ val infos = mutable.ArrayBuffer[ExecutorInfo]()
+ sc.addSparkListener(new SparkListener() {
+ override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded):
Unit = {
+ infos += executorAdded.executorInfo
+ }
+ })
+ val mockAddress = mock[RpcAddress]
+ Seq("1", "2").foreach { id =>
+ backend.driverEndpoint.askSync[Boolean](
+ RegisterExecutor(id, new MockExecutorRpcEndpointRef(conf),
mockAddress.host, 1,
+ Map(), Map(), Map.empty,
ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+ }
+ sc.listenerBus.waitUntilEmpty(executorUpTimeout.toMillis)
+ assert(infos.size === 2)
+ assert(infos.head.requestTime.isDefined)
+ assert(infos.last.requestTime.isEmpty)
+ }
+
private def testSubmitJob(sc: SparkContext, rdd: RDD[Int]): Unit = {
sc.submitJob(
rdd,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
