SPARK-2893: Do not swallow Exceptions when running a custom kryo regi…

…strator The previous behaviour of swallowing ClassNotFound exceptions when running a custom Kryo registrator could lead to difficult to debug problems later on at serialisation / deserialisation time, see SPARK-2878. Instead it is better to fail fast. Added test case. Author: Graham Dennis <[email protected]> Closes apache#1827 from GrahamDennis/feature/spark-2893 and squashes the following commits: fbe4cb6 [Graham Dennis] [SPARK-2878]: Update the test case to match the updated exception message 65e53c5 [Graham Dennis] [SPARK-2893]: Improve message when a spark.kryo.registrator fails. f480d85 [Graham Dennis] [SPARK-2893] Fix typo. b59d2c2 [Graham Dennis] SPARK-2893: Do not swallow Exceptions when running a custom spark.kryo.registrator
figo · Aug 14, 2014 · 6b8de0e · 6b8de0e
1 parent d069c5d
commit 6b8de0e
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 5 deletions.
diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -79,15 +79,16 @@ class KryoSerializer(conf: SparkConf)
     kryo.register(classOf[HttpBroadcast[_]], new KryoJavaSerializer())
 
     // Allow the user to register their own classes by setting spark.kryo.registrator
-    try {
-      for (regCls <- registrator) {
-        logDebug("Running user registrator: " + regCls)
+    for (regCls <- registrator) {
+      logDebug("Running user registrator: " + regCls)
+      try {
         val reg = Class.forName(regCls, true, classLoader).newInstance()
           .asInstanceOf[KryoRegistrator]
         reg.registerClasses(kryo)
+      } catch {
+        case e: Exception => 
+          throw new SparkException(s"Failed to invoke $regCls", e)
       }
-    } catch {
-      case e: Exception => logError("Failed to run spark.kryo.registrator", e)
     }
 
     // Register Chill's classes; we do this after our ranges and the user's own classes to let

diff --git a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
@@ -207,6 +207,16 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
         .fold(new ClassWithoutNoArgConstructor(10))((t1, t2) => new ClassWithoutNoArgConstructor(t1.x + t2.x)).x
     assert(10 + control.sum === result)
   }
+
+  test("kryo with nonexistent custom registrator should fail") {
+    import org.apache.spark.{SparkConf, SparkException}
+
+    val conf = new SparkConf(false)
+    conf.set("spark.kryo.registrator", "this.class.does.not.exist")
+
+    val thrown = intercept[SparkException](new KryoSerializer(conf).newInstance())
+    assert(thrown.getMessage.contains("Failed to invoke this.class.does.not.exist"))
+  }
 }
 
 class KryoSerializerResizableOutputSuite extends FunSuite {