Add fix for SPARK-7766.

Author: JoshRosen

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

@@ -177,6 +177,7 @@ private[spark] class KryoSerializerInstance(ks: KryoSerializer) extends Serializ
 
   override def serialize[T: ClassTag](t: T): ByteBuffer = {
     output.clear()
+    kryo.reset() // We must reset in case this serializer instance was reused (see SPARK-7766)
     try {
       kryo.writeClassAndObject(output, t)
     } catch {
@@ -202,6 +203,7 @@ private[spark] class KryoSerializerInstance(ks: KryoSerializer) extends Serializ
   }
 
   override def serializeStream(s: OutputStream): SerializationStream = {
+    kryo.reset() // We must reset in case this serializer instance was reused (see SPARK-7766)
     new KryoSerializationStream(kryo, s)
   }
 

core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala

@@ -345,6 +345,8 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
     assert (output1 === output2)
   }
 
+  // Regression test for SPARK-7766, an issue where disabling auto-reset and enabling
+  // reference-tracking would lead to corrupted output when serializer instances are re-used
   for (referenceTracking <- Set(true, false); autoReset <- Set(true, false)) {
     test(s"instance reuse with autoReset = $autoReset, referenceTracking = $referenceTracking") {
       testSerializerInstanceReuse(autoReset = autoReset, referenceTracking = referenceTracking)