You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2018/11/24 15:10:19 UTC
spark git commit: [SPARK-25786][CORE] If the ByteBuffer.hasArray is
false , it will throw UnsupportedOperationException for Kryo
Repository: spark
Updated Branches:
refs/heads/master de8489920 -> 7f5f7a967
[SPARK-25786][CORE] If the ByteBuffer.hasArray is false , it will throw UnsupportedOperationException for Kryo
## What changes were proposed in this pull request?
`deserialize` for kryo, the type of input parameter is ByteBuffer, if it is not backed by an accessible byte array. it will throw `UnsupportedOperationException`
Exception Info:
```
java.lang.UnsupportedOperationException was thrown.
java.lang.UnsupportedOperationException
at java.nio.ByteBuffer.array(ByteBuffer.java:994)
at org.apache.spark.serializer.KryoSerializerInstance.deserialize(KryoSerializer.scala:362)
```
## How was this patch tested?
Added a unit test
Closes #22779 from 10110346/InputStreamKryo.
Authored-by: liuxian <li...@zte.com.cn>
Signed-off-by: Sean Owen <se...@databricks.com>
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/7f5f7a96
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/7f5f7a96
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/7f5f7a96
Branch: refs/heads/master
Commit: 7f5f7a967d36d78f73d8fa1e178dfdb324d73bf1
Parents: de84899
Author: liuxian <li...@zte.com.cn>
Authored: Sat Nov 24 09:10:15 2018 -0600
Committer: Sean Owen <se...@databricks.com>
Committed: Sat Nov 24 09:10:15 2018 -0600
----------------------------------------------------------------------
.../apache/spark/serializer/KryoSerializer.scala | 16 +++++++++++++---
.../spark/serializer/KryoSerializerSuite.scala | 12 ++++++++++++
2 files changed, 25 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/7f5f7a96/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
index 66812a5..1e1c27c 100644
--- a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
+++ b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -42,7 +42,7 @@ import org.apache.spark.internal.Logging
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, SerializableConfiguration, SerializableJobConf, Utils}
+import org.apache.spark.util.{BoundedPriorityQueue, ByteBufferInputStream, SerializableConfiguration, SerializableJobConf, Utils}
import org.apache.spark.util.collection.CompactBuffer
/**
@@ -417,7 +417,12 @@ private[spark] class KryoSerializerInstance(
override def deserialize[T: ClassTag](bytes: ByteBuffer): T = {
val kryo = borrowKryo()
try {
- input.setBuffer(bytes.array(), bytes.arrayOffset() + bytes.position(), bytes.remaining())
+ if (bytes.hasArray) {
+ input.setBuffer(bytes.array(), bytes.arrayOffset() + bytes.position(), bytes.remaining())
+ } else {
+ input.setBuffer(new Array[Byte](4096))
+ input.setInputStream(new ByteBufferInputStream(bytes))
+ }
kryo.readClassAndObject(input).asInstanceOf[T]
} finally {
releaseKryo(kryo)
@@ -429,7 +434,12 @@ private[spark] class KryoSerializerInstance(
val oldClassLoader = kryo.getClassLoader
try {
kryo.setClassLoader(loader)
- input.setBuffer(bytes.array(), bytes.arrayOffset() + bytes.position(), bytes.remaining())
+ if (bytes.hasArray) {
+ input.setBuffer(bytes.array(), bytes.arrayOffset() + bytes.position(), bytes.remaining())
+ } else {
+ input.setBuffer(new Array[Byte](4096))
+ input.setInputStream(new ByteBufferInputStream(bytes))
+ }
kryo.readClassAndObject(input).asInstanceOf[T]
} finally {
kryo.setClassLoader(oldClassLoader)
http://git-wip-us.apache.org/repos/asf/spark/blob/7f5f7a96/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
index e413fe3..a7eed4b 100644
--- a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.serializer
import java.io.{ByteArrayInputStream, ByteArrayOutputStream, FileInputStream, FileOutputStream}
+import java.nio.ByteBuffer
import java.util.concurrent.Executors
import scala.collection.JavaConverters._
@@ -551,6 +552,17 @@ class KryoSerializerAutoResetDisabledSuite extends SparkFunSuite with SharedSpar
deserializationStream.close()
assert(serInstance.deserialize[Any](helloHello) === ((hello, hello)))
}
+
+ test("SPARK-25786: ByteBuffer.array -- UnsupportedOperationException") {
+ val serInstance = new KryoSerializer(conf).newInstance().asInstanceOf[KryoSerializerInstance]
+ val obj = "UnsupportedOperationException"
+ val serObj = serInstance.serialize(obj)
+ val byteBuffer = ByteBuffer.allocateDirect(serObj.array().length)
+ byteBuffer.put(serObj.array())
+ byteBuffer.flip()
+ assert(serInstance.deserialize[Any](serObj) === (obj))
+ assert(serInstance.deserialize[Any](byteBuffer) === (obj))
+ }
}
class ClassLoaderTestingObject
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org