[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