This is an automated email from the ASF dual-hosted git repository.
wangyum 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 34aecff1ee1d [SPARK-59674][CORE] Fix Comparison method violates its
general contract in FIFOSchedulingAlgorithm
34aecff1ee1d is described below
commit 34aecff1ee1d23d72c9da1cb9b3504084c8a1003
Author: Yuming Wang <[email protected]>
AuthorDate: Thu Oct 8 11:46:13 2026 +0800
[SPARK-59674][CORE] Fix Comparison method violates its general contract in
FIFOSchedulingAlgorithm
### What changes were proposed in this pull request?
`FIFOSchedulingAlgorithm.comparator` (used by `Pool.getSortedTaskSetQueue`
to order `TaskSetManager`s/`Pool`s in FIFO scheduling mode) compared `priority`
and `stageId` using:
```scala
math.signum(priority1 - priority2)
```
This subtraction can silently overflow when the two `Int` values are far
apart (e.g. span `Int.MinValue`..`Int.MaxValue`), which flips the sign of the
result and violates the `Comparator` contract (anti-symmetry/transitivity).
This PR replaces the subtraction-based `signum` comparisons with
`Integer.compare`, which compares the two values directly and cannot overflow:
```scala
var res = Integer.compare(s1.priority, s2.priority)
if (res == 0) {
res = Integer.compare(s1.stageId, s2.stageId)
}
```
### Why are the changes needed?
A broken comparator contract makes `java.util.TimSort` detect the
inconsistency during the merge phase and throw, which surfaces as a driver-side
failure when offering resources to executors:
```
java.lang.IllegalArgumentException: Comparison method violates its general
contract!
at java.util.TimSort.mergeHi(TimSort.java:903)
at java.util.TimSort.mergeAt(TimSort.java:520)
at java.util.TimSort.mergeForceCollapse(TimSort.java:461)
at java.util.TimSort.sort(TimSort.java:254)
at java.util.Arrays.sort(Arrays.java:1233)
at scala.collection.SeqLike.sorted(SeqLike.scala:659)
...
at
org.apache.spark.scheduler.Pool.getSortedTaskSetQueue(Pool.scala:109)
at
org.apache.spark.scheduler.TaskSchedulerImpl.resourceOffers(TaskSchedulerImpl.scala:440)
at
org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend$DriverEndpoint.makeOffers(...)
```
Once this is thrown from `resourceOffers`, resource offers stop being
processed correctly, which can stall task scheduling entirely.
`Integer.compare` performs a purely logical comparison and does not have this
overflow failure mode, so it satisfies the `Comparator` contract for any pair
of `Int` values.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Unit test.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5 (Anthropic)
Closes #58933 from wangyum/SPARK-59674.
Authored-by: Yuming Wang <[email protected]>
Signed-off-by: Yuming Wang <[email protected]>
(cherry picked from commit e5f07cb0546dff68cab7abaf8f1a260e5fe3dda8)
Signed-off-by: Yuming Wang <[email protected]>
---
.../spark/scheduler/SchedulingAlgorithm.scala | 8 +---
.../org/apache/spark/scheduler/PoolSuite.scala | 53 ++++++++++++++++++++--
2 files changed, 52 insertions(+), 9 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala
b/core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala
index 18ebbbe78a5b..a7db12dd97c5 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala
@@ -28,13 +28,9 @@ private[spark] trait SchedulingAlgorithm {
private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {
override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
- val priority1 = s1.priority
- val priority2 = s2.priority
- var res = math.signum(priority1 - priority2)
+ var res = Integer.compare(s1.priority, s2.priority)
if (res == 0) {
- val stageId1 = s1.stageId
- val stageId2 = s2.stageId
- res = math.signum(stageId1 - stageId2)
+ res = Integer.compare(s1.stageId, s2.stageId)
}
res < 0
}
diff --git a/core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala
index 37de8338ad90..1c18ac662faf 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala
@@ -39,12 +39,16 @@ class PoolSuite extends SparkFunSuite with
LocalSparkContext {
val APP_NAME = "PoolSuite"
val TEST_POOL = "testPool"
- def createTaskSetManager(stageId: Int, numTasks: Int, taskScheduler:
TaskSchedulerImpl)
- : TaskSetManager = {
+ def createTaskSetManager(
+ stageId: Int,
+ numTasks: Int,
+ taskScheduler: TaskSchedulerImpl,
+ priority: Int = 0,
+ stageAttemptId: Int = 0): TaskSetManager = {
val tasks = Array.tabulate[Task[_]](numTasks) { i =>
new FakeTask(stageId, i, Nil)
}
- new TaskSetManager(taskScheduler, new TaskSet(tasks, stageId, 0, 0, null,
+ new TaskSetManager(taskScheduler, new TaskSet(tasks, stageId,
stageAttemptId, priority, null,
ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID, None), 0)
}
@@ -407,6 +411,49 @@ class PoolSuite extends SparkFunSuite with
LocalSparkContext {
s"and '$expectedBatchPrefix'.\nCaptured logs:\n${logs.mkString("\n")}")
}
+ test("SPARK-59674: FIFO orders priorities and stage ids at Int bounds") {
+ sc = new SparkContext(LOCAL, APP_NAME)
+ val taskScheduler = new TaskSchedulerImpl(sc)
+ // This sequence makes getSortedTaskSetQueue throw
IllegalArgumentException:
+ // Comparison method violates its general contract!
+ // when priority or stageId is ordered with signum(a - b).
+ val boundaryValues = Array(
+ 0, Int.MaxValue, 0, 0, Int.MaxValue, Int.MinValue, Int.MinValue,
+ Int.MaxValue, Int.MaxValue, 0,
+ 0, Int.MaxValue, Int.MaxValue, 0, 0, 0, Int.MaxValue, Int.MaxValue,
+ Int.MinValue, Int.MaxValue, 0, 0, 0, Int.MinValue, Int.MinValue,
Int.MinValue,
+ Int.MinValue, Int.MaxValue, 0, 0, 0, Int.MinValue, Int.MinValue,
Int.MinValue,
+ Int.MaxValue, 0, Int.MinValue, Int.MinValue, Int.MaxValue, Int.MaxValue,
+ Int.MinValue, Int.MinValue,
+ Int.MaxValue, 0, Int.MinValue, Int.MinValue, Int.MaxValue, Int.MaxValue,
+ Int.MinValue, Int.MinValue)
+ val expected = boundaryValues.sorted.toSeq
+
+ // Equal stage ids so the priority comparison is the only one that can
overflow.
+ val priorityPool = new Pool("", FIFO, 0, 0)
+ boundaryValues.zipWithIndex.foreach { case (priority, attempt) =>
+ priorityPool.addSchedulable(createTaskSetManager(
+ stageId = 0,
+ numTasks = 1,
+ taskScheduler,
+ priority = priority,
+ stageAttemptId = attempt))
+ }
+ assert(priorityPool.getSortedTaskSetQueue.map(_.priority) === expected)
+
+ // Equal priorities so ordering falls through to stageId.
+ val stagePool = new Pool("", FIFO, 0, 0)
+ boundaryValues.zipWithIndex.foreach { case (stageId, attempt) =>
+ stagePool.addSchedulable(createTaskSetManager(
+ stageId,
+ numTasks = 1,
+ taskScheduler,
+ priority = 1,
+ stageAttemptId = attempt))
+ }
+ assert(stagePool.getSortedTaskSetQueue.map(_.stageId) === expected)
+ }
+
private def verifyPool(rootPool: Pool, poolName: String,
expectedInitMinShare: Int,
expectedInitWeight: Int, expectedSchedulingMode:
SchedulingMode): Unit = {
val selectedPool = rootPool.getSchedulableByName(poolName)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
