SPARK-5949 HighlyCompressedMapStatus needs more classes registered w/ kryo

Author: squito

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

@@ -24,6 +24,8 @@ import com.esotericsoftware.kryo.{Kryo, KryoException}
 import com.esotericsoftware.kryo.io.{Input => KryoInput, Output => KryoOutput}
 import com.esotericsoftware.kryo.serializers.{JavaSerializer => KryoJavaSerializer}
 import com.twitter.chill.{AllScalaRegistrar, EmptyScalaKryoInstantiator}
+import org.roaringbitmap.{ArrayContainer, RoaringArray, RoaringBitmap}
+
 
 import org.apache.spark._
 import org.apache.spark.api.python.PythonBroadcast
@@ -202,6 +204,12 @@ private[serializer] object KryoSerializer {
     classOf[GetBlock],
     classOf[CompressedMapStatus],
     classOf[HighlyCompressedMapStatus],
+    classOf[RoaringBitmap],
+    classOf[RoaringArray],
+    classOf[RoaringArray.Element],
+    classOf[Array[RoaringArray.Element]],
+    classOf[ArrayContainer],
+    classOf[Array[Short]],
     classOf[CompactBuffer[_]],
     classOf[BlockManagerId],
     classOf[Array[Byte]],

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

@@ -17,6 +17,9 @@
 
 package org.apache.spark.serializer
 
+import org.apache.spark.scheduler.HighlyCompressedMapStatus
+import org.apache.spark.storage.BlockManagerId
+
 import scala.collection.mutable
 import scala.reflect.ClassTag
 
@@ -242,6 +245,16 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
       ser.newInstance().deserialize[ClassLoaderTestingObject](bytes)
     }
   }
+
+  test("registration of HighlyCompressedMapStatus") {
+    val conf = new SparkConf(false)
+    conf.set("spark.kryo.registrationRequired", "true")
+    val hcmo = HighlyCompressedMapStatus(BlockManagerId("exec-1", "host", 1234), Array(0l,2l,5l))
+    val ser = new KryoSerializer(conf)
+    val serInstance = ser.newInstance()
+    serInstance.serialize(hcmo)
+
+  }
 }