[SPARK-59674][CORE] Fix Comparison method violates its general contract in FIFOSchedulingAlgorithm

[wangyum]

What changes were proposed in this pull request?

FIFOSchedulingAlgorithm.comparator (used by Pool.getSortedTaskSetQueue to order TaskSetManagers/Pools in FIFO scheduling mode) compared priority and stageId using:

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:

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?

Manually verified both the failure and the fix with a minimal reproduction outside the Spark test suite:

import java.util.{ArrayList, Comparator}

val buggyComparator = new Comparator[Integer]() {
  override def compare(a: Integer, b: Integer): Int = math.signum(a - b)
}

val values: Array[Int] = 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 list1 = new ArrayList[Integer]()
values.foreach(v => list1.add(Int.box(v)))
list1.sort(buggyComparator)

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Sonnet 5 (Anthropic)

[wangyum]

cc @cloud-fan

[cloud-fan]

Review summary

The overflow-safe implementation is sound and preserves the intended FIFO ordering. The remaining issue is durable regression coverage: the current Spark tests do not distinguish either Integer.compare change from the previous subtraction, so both boundary paths should be covered in-repo. This is a non-blocking test gap, not a production-code correctness issue.

Findings

1 total: 0 P0, 0 P1, 1 P2, 0 P3.

Non-blocking (P2)

• Add regression coverage for both overflow comparisons — core/src/main/scala/org/apache/spark/scheduler/SchedulingAlgorithm.scala:31 — see inline.

[cloud-fan]

Non-blocking (P2): Please add an in-repo regression test that fails against the old subtraction for both changed comparisons: priorities spanning Int.MinValue and Int.MaxValue, and equal-priority schedulables whose stage IDs span the same bounds. PoolSuite currently fixes priority at 0 and uses only small stage IDs, so it passes with the old comparator and would not catch a reintroduction of this scheduler-stalling failure.

[cloud-fan]

Thanks for the ping. I reviewed the current patch and will post the review outcome here.