[SPARK-42071][CORE] Register scala.math.Ordering$Reverse to KyroSerializer

This PR aims to register `scala.math.Ordering$Reverse` to KyroSerializer.

Scala 2.12.12 added a new class 'Reverse' via scala/scala#8965. This affects Apache Spark 3.2.0+.

No.

Pass the CIs with newly added test case.

Closes apache#39578 from dongjoon-hyun/SPARK-42071.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit e3c0fbe)
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala

@@ -219,6 +219,7 @@ class KryoSerializer(conf: SparkConf)
     kryo.register(Nil.getClass)
     kryo.register(Utils.classForName("scala.collection.immutable.$colon$colon"))
     kryo.register(Utils.classForName("scala.collection.immutable.Map$EmptyMap$"))
+    kryo.register(Utils.classForName("scala.math.Ordering$Reverse"))
     kryo.register(classOf[ArrayBuffer[Any]])
 
     // We can't load those class directly in order to avoid unnecessary jar dependencies.

core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala

@@ -38,7 +38,7 @@ import org.apache.spark.internal.io.FileCommitProtocol.TaskCommitMessage
 import org.apache.spark.scheduler.HighlyCompressedMapStatus
 import org.apache.spark.serializer.KryoTest._
 import org.apache.spark.storage.BlockManagerId
-import org.apache.spark.util.ThreadUtils
+import org.apache.spark.util.{ThreadUtils, Utils}
 
 class KryoSerializerSuite extends SparkFunSuite with SharedSparkContext {
   conf.set(SERIALIZER, "org.apache.spark.serializer.KryoSerializer")
@@ -171,6 +171,18 @@ class KryoSerializerSuite extends SparkFunSuite with SharedSparkContext {
       mutable.HashMap(1 -> "one", 2 -> "two", 3 -> "three")))
   }
 
+  test("SPARK-42071: Register scala.math.Ordering$Reverse") {
+    val conf = new SparkConf(false)
+    conf.set(KRYO_REGISTRATION_REQUIRED, true)
+
+    val ser = new KryoSerializer(conf).newInstance()
+    def check[T: ClassTag](t: T): Unit = {
+      assert(ser.deserialize[T](ser.serialize(t)) === t)
+    }
+    // Scala 2.12.12 added a new class 'Reverse' via https://github.com/scala/scala/pull/8965
+    check(Utils.classForName("scala.math.Ordering$Reverse"))
+  }
+
   test("Bug: SPARK-10251") {
     val ser = new KryoSerializer(conf.clone.set(KRYO_REGISTRATION_REQUIRED, true))
       .newInstance()