+ bug fixes;

test added

Conflicts:

	core/src/test/scala/spark/KryoSerializerSuite.scala

2 files changed, 20 insertions, 8 deletions

diff --git a/core/src/main/scala/spark/rdd/ParallelCollectionRDD.scala b/core/src/main/scala/spark/rdd/ParallelCollectionRDD.scala
index 104257ac07..dfeb7be077 100644
--- a/core/src/main/scala/spark/rdd/ParallelCollectionRDD.scala
+++ b/core/src/main/scala/spark/rdd/ParallelCollectionRDD.scala
@@ -21,14 +21,18 @@ import scala.collection.immutable.NumericRange
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.Map
 import spark._
-import java.io.{ObjectInput, ObjectOutput, Externalizable}
+import java.io._
 import java.nio.ByteBuffer
+import scala.Serializable
 
 private[spark] class ParallelCollectionPartition[T: ClassManifest](
                                                                     var rddId: Long,
                                                                     var slice: Int,
                                                                     var values: Seq[T])
-  extends Partition with Externalizable {
+  extends Partition with Serializable {
+
+  // for externalization
+  def this() = this(0, 0, null)
 
   def iterator: Iterator[T] = values.iterator
 
@@ -41,7 +45,8 @@ private[spark] class ParallelCollectionPartition[T: ClassManifest](
 
   override def index: Int = slice
 
-  override def writeExternal(out: ObjectOutput) {
+  @throws(classOf[IOException])
+  private def writeObject(out: ObjectOutputStream): Unit = {
     out.writeLong(rddId)
     out.writeInt(slice)
     out.writeInt(values.size)
@@ -59,22 +64,23 @@ private[spark] class ParallelCollectionPartition[T: ClassManifest](
     })
   }
 
-  override def readExternal(in: ObjectInput) {
+  @throws(classOf[IOException])
+  private def readObject(in: ObjectInputStream): Unit = {
     rddId = in.readLong()
     slice = in.readInt()
     val s = in.readInt()
     val ser = SparkEnv.get.serializer.newInstance()
     var bb = ByteBuffer.allocate(1024)
-    values = (0 until s).map({
+    values = (0 until s).map(i => {
       val s = in.readInt()
       if (bb.capacity() < s) {
         bb = ByteBuffer.allocate(s)
       } else {
         bb.clear()
       }
-      in.readFully(bb.array())
+      in.readFully(bb.array(), 0, s)
       bb.limit(s)
-      ser.deserialize(bb)
+      ser.deserialize(bb): T
     }).toSeq
   }
 }
diff --git a/core/src/test/scala/spark/KryoSerializerSuite.scala b/core/src/test/scala/spark/KryoSerializerSuite.scala
index 793b0b66c4..5ae7a427cc 100644
--- a/core/src/test/scala/spark/KryoSerializerSuite.scala
+++ b/core/src/test/scala/spark/KryoSerializerSuite.scala
@@ -24,7 +24,7 @@ import org.scalatest.FunSuite
 import com.esotericsoftware.kryo._
 
 import SparkContext._
-import spark.test.{ClassWithoutNoArgConstructor, MyRegistrator}
+import spark.test._
 
 class KryoSerializerSuite extends FunSuite with SharedSparkContext {
   test("basic types") {
@@ -134,6 +134,12 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
     assert(control == result.toSeq)
   }
 
+  test("kryo with parallelize") {
+    val control = 1 :: 2 :: Nil
+    val result = sc.parallelize(control.map(new ClassWithoutNoArgConstructor(_))).map(_.x).collect()
+    assert (control == result.toSeq)
+  }
+
   override def beforeAll() {
     System.setProperty("spark.serializer", "spark.KryoSerializer")
     System.setProperty("spark.kryo.registrator", classOf[MyRegistrator].getName)