make default for kryo max buffer size 16MB

Author: koertkuipers

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

@@ -39,12 +39,12 @@ class KryoSerializer(conf: SparkConf)
   with Logging
   with Serializable {
 
-  private val bufferSizeMb = conf.getInt("spark.kryoserializer.buffer.mb", 2)
-  private val maxBufferSizeMb = conf.getInt("spark.kryoserializer.buffer.max.mb", bufferSizeMb)
+  private val bufferSize = conf.getInt("spark.kryoserializer.buffer.mb", 2) * 1024 * 1024
+  private val maxBufferSize = conf.getInt("spark.kryoserializer.buffer.max.mb", 16) * 1024 * 1024
   private val referenceTracking = conf.getBoolean("spark.kryo.referenceTracking", true)
   private val registrator = conf.getOption("spark.kryo.registrator")
 
-  def newKryoOutput() = new KryoOutput(bufferSizeMb * 1024 * 1024, maxBufferSizeMb * 1024 * 1024)
+  def newKryoOutput() = new KryoOutput(bufferSize, math.max(bufferSize, maxBufferSize))
 
   def newKryo(): Kryo = {
     val instantiator = new EmptyScalaKryoInstantiator

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

@@ -206,6 +206,7 @@ class KryoSerializerResizableOutputSuite extends FunSuite {
     val conf = new SparkConf(false)
     conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
     conf.set("spark.kryoserializer.buffer.mb", "1")
+    conf.set("spark.kryoserializer.buffer.max.mb", "1")
     val sc = new SparkContext("local", "test", conf)
     intercept[SparkException](sc.parallelize(x).collect)
     LocalSparkContext.stop(sc)