@@ -21,39 +21,40 @@ import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
2121
2222import scala .util .Random
2323
24- import org .scalatest .FunSuite
24+ import org .scalatest .{ Assertions , FunSuite }
2525
2626import org .apache .spark .SparkConf
2727import org .apache .spark .serializer .KryoTest .RegistratorWithoutAutoReset
2828
29- private case class MyCaseClass (foo : Int , bar : String )
3029
3130class SerializerPropertiesSuite extends FunSuite {
3231
32+ import SerializerPropertiesSuite ._
33+
3334 test(" JavaSerializer does not support relocation" ) {
34- testSupportsRelocationOfSerializedObjects(new JavaSerializer (new SparkConf ()))
35+ val ser = new JavaSerializer (new SparkConf ())
36+ testSupportsRelocationOfSerializedObjects(ser, generateRandomItem)
3537 }
3638
3739 test(" KryoSerializer supports relocation when auto-reset is enabled" ) {
3840 val ser = new KryoSerializer (new SparkConf )
3941 assert(ser.newInstance().asInstanceOf [KryoSerializerInstance ].getAutoReset())
40- testSupportsRelocationOfSerializedObjects(ser)
42+ testSupportsRelocationOfSerializedObjects(ser, generateRandomItem )
4143 }
4244
4345 test(" KryoSerializer does not support relocation when auto-reset is disabled" ) {
4446 val conf = new SparkConf ().set(" spark.kryo.registrator" ,
4547 classOf [RegistratorWithoutAutoReset ].getName)
4648 val ser = new KryoSerializer (conf)
4749 assert(! ser.newInstance().asInstanceOf [KryoSerializerInstance ].getAutoReset())
48- testSupportsRelocationOfSerializedObjects(ser)
50+ testSupportsRelocationOfSerializedObjects(ser, generateRandomItem )
4951 }
5052
51- def testSupportsRelocationOfSerializedObjects (serializer : Serializer ): Unit = {
52- val NUM_TRIALS = 100
53- if (! serializer.supportsRelocationOfSerializedObjects) {
54- return
55- }
56- val rand = new Random (42 )
53+ }
54+
55+ object SerializerPropertiesSuite extends Assertions {
56+
57+ def generateRandomItem (rand : Random ): Any = {
5758 val randomFunctions : Seq [() => Any ] = Seq (
5859 () => rand.nextInt(),
5960 () => rand.nextString(rand.nextInt(10 )),
@@ -66,14 +67,21 @@ class SerializerPropertiesSuite extends FunSuite {
6667 (x, x)
6768 }
6869 )
69- def generateRandomItem (): Any = {
70- randomFunctions(rand.nextInt(randomFunctions.size)).apply()
71- }
70+ randomFunctions(rand.nextInt(randomFunctions.size)).apply()
71+ }
7272
73+ def testSupportsRelocationOfSerializedObjects (
74+ serializer : Serializer ,
75+ generateRandomItem : Random => Any ): Unit = {
76+ if (! serializer.supportsRelocationOfSerializedObjects) {
77+ return
78+ }
79+ val NUM_TRIALS = 10
80+ val rand = new Random (42 )
7381 for (_ <- 1 to NUM_TRIALS ) {
7482 val items = {
7583 // Make sure that we have duplicate occurrences of the same object in the stream:
76- val randomItems = Seq .fill(10 )(generateRandomItem())
84+ val randomItems = Seq .fill(10 )(generateRandomItem(rand ))
7785 randomItems ++ randomItems.take(5 )
7886 }
7987 val baos = new ByteArrayOutputStream ()
@@ -99,5 +107,6 @@ class SerializerPropertiesSuite extends FunSuite {
99107 deserializedItemsStream.close()
100108 }
101109 }
102-
103110}
111+
112+ private case class MyCaseClass (foo : Int , bar : String )
0 commit comments