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 9d4168f8ffc9 [SPARK-58886][CORE] Fix Int overflow in
`CoarseGrainedSchedulerBackend.requestExecutors`
9d4168f8ffc9 is described below
commit 9d4168f8ffc924dded4f9c95348cbe7dea362eea
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Aug 19 16:42:47 2026 -0700
[SPARK-58886][CORE] Fix Int overflow in
`CoarseGrainedSchedulerBackend.requestExecutors`
### What changes were proposed in this pull request?
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.
### Why are the changes needed?
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.
### Does this PR introduce _any_ user-facing change?
No. It only prevents a pathological bookkeeping value in the corner case
above.
### How was this patch tested?
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.
### Was this patch authored or co-authored using generative AI tooling?
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 8871a4b5b880..e07cde82ffa0 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
@@ -794,6 +794,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
@@ -812,9 +817,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 10eb884619b5..8c317661ef8a 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)
+ }
+
test("UpdateUserCredentials is broadcast to all registered executors") {
val conf = new SparkConf()
.setMaster("local-cluster[0, 3, 1024]")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
