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

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 #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

Author: GrahamDennis

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

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 {