Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-7392][Core] bugfix: Kryo buffer size cannot be larger than 2M #5934

Closed
wants to merge 4 commits into from
Closed

[SPARK-7392][Core] bugfix: Kryo buffer size cannot be larger than 2M #5934

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d91e5ed
change kb to mb
liyezhang556520 May 6, 2015
9bf93e9
add tests
liyezhang556520 May 6, 2015
8693288
replace multiplier with ByteUnit methods
liyezhang556520 May 6, 2015
5707e04
fix import order
liyezhang556520 May 6, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import org.apache.spark._
import org.apache.spark.api.python.PythonBroadcast
import org.apache.spark.broadcast.HttpBroadcast
import org.apache.spark.network.nio.{GetBlock, GotBlock, PutBlock}
import org.apache.spark.network.util.ByteUnit
import org.apache.spark.scheduler.{CompressedMapStatus, HighlyCompressedMapStatus}
import org.apache.spark.storage._
import org.apache.spark.util.BoundedPriorityQueue
Expand All @@ -51,18 +52,18 @@ class KryoSerializer(conf: SparkConf)

private val bufferSizeKb = conf.getSizeAsKb("spark.kryoserializer.buffer", "64k")

if (bufferSizeKb >= 2048) {
if (bufferSizeKb >= ByteUnit.GiB.toKiB(2)) {
throw new IllegalArgumentException("spark.kryoserializer.buffer must be less than " +
s"2048 mb, got: + $bufferSizeKb mb.")
s"2048 mb, got: + ${ByteUnit.KiB.toMiB(bufferSizeKb)} mb.")
}
private val bufferSize = (bufferSizeKb * 1024).toInt
private val bufferSize = ByteUnit.KiB.toBytes(bufferSizeKb).toInt

val maxBufferSizeMb = conf.getSizeAsMb("spark.kryoserializer.buffer.max", "64m").toInt
if (maxBufferSizeMb >= 2048) {
if (maxBufferSizeMb >= ByteUnit.GiB.toMiB(2)) {
throw new IllegalArgumentException("spark.kryoserializer.buffer.max must be less than " +
s"2048 mb, got: + $maxBufferSizeMb mb.")
}
private val maxBufferSize = maxBufferSizeMb * 1024 * 1024
private val maxBufferSize = ByteUnit.MiB.toBytes(maxBufferSizeMb).toInt

private val referenceTracking = conf.getBoolean("spark.kryo.referenceTracking", true)
private val registrationRequired = conf.getBoolean("spark.kryo.registrationRequired", false)
Expand Down
30 changes: 30 additions & 0 deletions core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,36 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.set("spark.kryo.registrator", classOf[MyRegistrator].getName)

test("configuration limits") {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

General style note: this test has a lot of repetition in it; the fact that we have four conf variables, conf1, conf2, conf3, and conf4, suggests that this could have benefitted from an inline def that abstracts away the actual configuration manipulation and KryoSerializer instantiation. Either this or whitespace between the individual cases would make this test easier to quickly scan through.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, it would have been nice to have the test name reference the JIRA number.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I'll make a new PR.

val conf1 = conf.clone()
val kryoBufferProperty = "spark.kryoserializer.buffer"
val kryoBufferMaxProperty = "spark.kryoserializer.buffer.max"
conf1.set(kryoBufferProperty, "64k")
conf1.set(kryoBufferMaxProperty, "64m")
new KryoSerializer(conf1).newInstance()
// 2048m = 2097152k
conf1.set(kryoBufferProperty, "2097151k")
conf1.set(kryoBufferMaxProperty, "64m")
// should not throw exception when kryoBufferMaxProperty < kryoBufferProperty
new KryoSerializer(conf1).newInstance()
conf1.set(kryoBufferMaxProperty, "2097151k")
new KryoSerializer(conf1).newInstance()
val conf2 = conf.clone()
conf2.set(kryoBufferProperty, "2048m")
val thrown1 = intercept[IllegalArgumentException](new KryoSerializer(conf2).newInstance())
assert(thrown1.getMessage.contains(kryoBufferProperty))
val conf3 = conf.clone()
conf3.set(kryoBufferMaxProperty, "2048m")
val thrown2 = intercept[IllegalArgumentException](new KryoSerializer(conf3).newInstance())
assert(thrown2.getMessage.contains(kryoBufferMaxProperty))
val conf4 = conf.clone()
conf4.set(kryoBufferProperty, "2g")
conf4.set(kryoBufferMaxProperty, "3g")
val thrown3 = intercept[IllegalArgumentException](new KryoSerializer(conf4).newInstance())
assert(thrown3.getMessage.contains(kryoBufferProperty))
assert(!thrown3.getMessage.contains(kryoBufferMaxProperty))
}

test("basic types") {
val ser = new KryoSerializer(conf).newInstance()
def check[T: ClassTag](t: T) {
Expand Down