allow non-Spark classes

Author: pjfanning

core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

@@ -920,14 +920,10 @@ private[spark] class JsonProtocol(sparkConf: SparkConf) extends JsonUtils {
       case `blockUpdate` => blockUpdateFromJson(json)
       case `resourceProfileAdded` => resourceProfileAddedFromJson(json)
       case other =>
-        if (other.startsWith("org.apache.spark")) {
-          val otherClass = Utils.classForName(other)
-          if (classOf[SparkListenerEvent].isAssignableFrom(otherClass)) {
-            mapper.readValue(json.toString, otherClass)
-              .asInstanceOf[SparkListenerEvent]
-          } else {
-            throw new SparkException(s"Unknown event type: $other")
-          }
+        val otherClass = Utils.classForName(other)
+        if (classOf[SparkListenerEvent].isAssignableFrom(otherClass)) {
+          mapper.readValue(json.toString, otherClass)
+            .asInstanceOf[SparkListenerEvent]
         } else {
           throw new SparkException(s"Unknown event type: $other")
         }

core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala

@@ -1023,18 +1023,17 @@ class jsonProtocolSuite extends SparkFunSuite {
     ))
   }
 
-  test("SPARK-52381: only read Spark classes") {
+  test("SPARK-52381: handle class not found") {
     val unknownJson =
       """{
         |  "Event" : "com.example.UnknownEvent",
         |  "foo" : "foo"
         |}""".stripMargin
     try {
       jsonProtocol.sparkEventFromJson(unknownJson)
-      fail("Expected SparkException for unknown event type")
+      fail("Expected ClassNotFoundException for unknown event type")
     } catch {
-      case e: SparkException =>
-        assert(e.getMessage.startsWith("Unknown event type"))
+      case e: ClassNotFoundException =>
     }
   }