Register PythonBroadcast with Kryo.

Author: JoshRosen

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

@@ -26,6 +26,7 @@ import com.esotericsoftware.kryo.serializers.{JavaSerializer => KryoJavaSerializ
 import com.twitter.chill.{AllScalaRegistrar, EmptyScalaKryoInstantiator}
 
 import org.apache.spark._
+import org.apache.spark.api.python.PythonBroadcast
 import org.apache.spark.broadcast.HttpBroadcast
 import org.apache.spark.network.nio.{PutBlock, GotBlock, GetBlock}
 import org.apache.spark.scheduler.MapStatus
@@ -90,6 +91,7 @@ class KryoSerializer(conf: SparkConf)
     // Allow sending SerializableWritable
     kryo.register(classOf[SerializableWritable[_]], new KryoJavaSerializer())
     kryo.register(classOf[HttpBroadcast[_]], new KryoJavaSerializer())
+    kryo.register(classOf[PythonBroadcast], new KryoJavaSerializer())
 
     try {
       // Use the default classloader when calling the user registrator.

core/src/test/scala/org/apache/spark/api/python/PythonBroadcastSuite.scala

@@ -23,12 +23,13 @@ import java.io.{PrintWriter, File}
 
 import org.scalatest.{Matchers, FunSuite}
 
-import org.apache.spark.SparkConf
+import org.apache.spark.{SharedSparkContext, SparkConf}
 import org.apache.spark.serializer.KryoSerializer
 import org.apache.spark.util.Utils
 
-
-class PythonBroadcastSuite extends FunSuite with Matchers {
+// This test suite uses SharedSparkContext because we need a SparkEnv in order to deserialize
+// a PythonBroadcast:
+class PythonBroadcastSuite extends FunSuite with Matchers with SharedSparkContext {
   test("PythonBroadcast can be serialized with Kryo (SPARK-4882)") {
     val tempDir = Utils.createTempDir()
     val broadcastedString = "Hello, world!"