[SPARK-18617][CORE][STREAMING] Close "kryo auto pick" feature for Spark Streaming #16052

uncleGen · 2016-11-29T06:14:20Z

What changes were proposed in this pull request?

#15992 provided a solution to fix the bug, i.e. receiver data can not be deserialized properly. As @zsxwing said, it is a critical bug, but we should not break APIs between maintenance releases. It may be a rational choice to close auto pick kryo serializer for Spark Streaming in the first step. I will continue #15992 to optimize the solution.

How was this patch tested?

existing ut

uncleGen · 2016-11-29T06:18:24Z

cc @zsxwing

SparkQA · 2016-11-29T06:22:56Z

Test build #69300 has finished for PR 16052 at commit e4416bd.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-11-29T06:47:38Z

Test build #69303 has started for PR 16052 at commit d205ebd.

rxin · 2016-11-29T07:41:44Z

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

@@ -77,8 +77,8 @@ private[spark] class SerializerManager(defaultSerializer: Serializer, conf: Spar
    primitiveAndPrimitiveArrayClassTags.contains(ct) || ct == stringClassTag
  }

-  def getSerializer(ct: ClassTag[_]): Serializer = {
-    if (canUseKryo(ct)) {
+  def getSerializer(ct: ClassTag[_], autoPick: Boolean): Serializer = {


document what this means

srowen · 2016-11-29T10:08:17Z

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

@@ -155,7 +155,12 @@ private[spark] class SerializerManager(defaultSerializer: Serializer, conf: Spar
      outputStream: OutputStream,
      values: Iterator[T]): Unit = {
    val byteStream = new BufferedOutputStream(outputStream)
-    val ser = getSerializer(implicitly[ClassTag[T]]).newInstance()
+    val ser = blockId match {


Is this simpler as ...

val autoPick = blockId.isInstanceOf[StreamBlockId] val ser = getSerializer(implicitly[ClassTag[T]], autoPick).newInstance()

to avoid duplication?

Ah, yours is better.

SparkQA · 2016-11-29T13:54:33Z

Test build #69323 has finished for PR 16052 at commit cd7e595.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2016-11-30T00:20:40Z

@uncleGen Could you add the following test to StreamingContextSuite? Otherwise, LGTM.

  test("SPARK-18560 Receiver data should be deserialized properly.") {
    // Start a two nodes cluster, so receiver will use one node, and Spark jobs will use the
    // other one. Then Spark jobs need to fetch remote blocks and it will trigger SPARK-18560.
    val conf = new SparkConf().setMaster("local-cluster[2,1,1024]").setAppName(appName)
    sc = new SparkContext(conf)
    ssc = new StreamingContext(sc, Milliseconds(100))
    val input = ssc.receiverStream(new TestReceiver)
    input.count().foreachRDD { rdd =>
      // Make sure we can read from BlockRDD
      if (rdd.collect().headOption.getOrElse(0L) > 0) {
        // Stop StreamingContext to unblock "awaitTerminationOrTimeout"
        new Thread() {
          setDaemon(true)
          override def run(): Unit = {
            ssc.stop(stopSparkContext = true, stopGracefully = true)
          }
        }.start()
      }
    }
    ssc.start()
    ssc.awaitTerminationOrTimeout(60000)
  }

uncleGen · 2016-11-30T02:13:56Z

OK

SparkQA · 2016-11-30T07:38:37Z

Test build #69380 has finished for PR 16052 at commit 39b4867.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class FakeByteArrayReceiver extends Receiver[Array[Byte]](StorageLevel.MEMORY_ONLY) with Logging

rxin · 2016-11-30T07:44:40Z

Merging in master/branch-2.1.

…rk Streaming ## What changes were proposed in this pull request? #15992 provided a solution to fix the bug, i.e. **receiver data can not be deserialized properly**. As zsxwing said, it is a critical bug, but we should not break APIs between maintenance releases. It may be a rational choice to close auto pick kryo serializer for Spark Streaming in the first step. I will continue #15992 to optimize the solution. ## How was this patch tested? existing ut Author: uncleGen <[email protected]> Closes #16052 from uncleGen/SPARK-18617. (cherry picked from commit 56c82ed) Signed-off-by: Reynold Xin <[email protected]>

rxin · 2016-11-30T07:45:57Z

This doesn't merge cleanly into branch-2.0. @uncleGen can you submit a pull request for branch-2.0?

zsxwing · 2016-11-30T07:48:38Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

@@ -869,6 +891,31 @@ object TestReceiver {
  val counter = new AtomicInteger(1)
 }

+class FakeByteArrayReceiver extends Receiver[Array[Byte]](StorageLevel.MEMORY_ONLY) with Logging {


nit: Why create a new class? Is there any concern to just use TestReceiver?

@zsxwing yes, failure occurs when receiver store Array[Byte] data and the automatic serializer selection would pick JavaSerializer. However, after get from remote executor, the input-stream data will be deserialized with KryoSerializer, leading to the com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 13994

BTW, existing unit test could cover other cases besides Array[Byte] type.

Any class in primitiveAndPrimitiveArrayClassTags can trigger this issue. That's why you can also use the existing TestReceiver. It's Receiver[Int].

uncleGen · 2016-11-30T08:46:22Z

@rxin OK, I will backport it to branch-2.0

…e. Receiver data should be deserialized properly ## What changes were proposed in this pull request? Fixed the potential SparkContext leak in `StreamingContextSuite.SPARK-18560 Receiver data should be deserialized properly` which was added in #16052. I also removed FakeByteArrayReceiver and used TestReceiver directly. ## How was this patch tested? Jenkins Author: Shixiong Zhu <[email protected]> Closes #16091 from zsxwing/SPARK-18617-follow-up. (cherry picked from commit 0a81121) Signed-off-by: Reynold Xin <[email protected]>

…e. Receiver data should be deserialized properly ## What changes were proposed in this pull request? Fixed the potential SparkContext leak in `StreamingContextSuite.SPARK-18560 Receiver data should be deserialized properly` which was added in #16052. I also removed FakeByteArrayReceiver and used TestReceiver directly. ## How was this patch tested? Jenkins Author: Shixiong Zhu <[email protected]> Closes #16091 from zsxwing/SPARK-18617-follow-up.

…e for Spark Streaming ## What changes were proposed in this pull request? This is a follow-up PR to backport #16052 to branch-2.0 with incremental update in #16091 ## How was this patch tested? new unit test cc zsxwing rxin Author: uncleGen <[email protected]> Closes #16096 from uncleGen/branch-2.0-SPARK-18617.

…rk Streaming ## What changes were proposed in this pull request? apache#15992 provided a solution to fix the bug, i.e. **receiver data can not be deserialized properly**. As zsxwing said, it is a critical bug, but we should not break APIs between maintenance releases. It may be a rational choice to close auto pick kryo serializer for Spark Streaming in the first step. I will continue apache#15992 to optimize the solution. ## How was this patch tested? existing ut Author: uncleGen <[email protected]> Closes apache#16052 from uncleGen/SPARK-18617.

…e. Receiver data should be deserialized properly ## What changes were proposed in this pull request? Fixed the potential SparkContext leak in `StreamingContextSuite.SPARK-18560 Receiver data should be deserialized properly` which was added in apache#16052. I also removed FakeByteArrayReceiver and used TestReceiver directly. ## How was this patch tested? Jenkins Author: Shixiong Zhu <[email protected]> Closes apache#16091 from zsxwing/SPARK-18617-follow-up.

…rk Streaming ## What changes were proposed in this pull request? apache#15992 provided a solution to fix the bug, i.e. **receiver data can not be deserialized properly**. As zsxwing said, it is a critical bug, but we should not break APIs between maintenance releases. It may be a rational choice to close auto pick kryo serializer for Spark Streaming in the first step. I will continue apache#15992 to optimize the solution. ## How was this patch tested? existing ut Author: uncleGen <[email protected]> Closes apache#16052 from uncleGen/SPARK-18617.

…e. Receiver data should be deserialized properly ## What changes were proposed in this pull request? Fixed the potential SparkContext leak in `StreamingContextSuite.SPARK-18560 Receiver data should be deserialized properly` which was added in apache#16052. I also removed FakeByteArrayReceiver and used TestReceiver directly. ## How was this patch tested? Jenkins Author: Shixiong Zhu <[email protected]> Closes apache#16091 from zsxwing/SPARK-18617-follow-up.

…rk Streaming ## What changes were proposed in this pull request? apache#15992 provided a solution to fix the bug, i.e. **receiver data can not be deserialized properly**. As zsxwing said, it is a critical bug, but we should not break APIs between maintenance releases. It may be a rational choice to close auto pick kryo serializer for Spark Streaming in the first step. I will continue apache#15992 to optimize the solution. ## How was this patch tested? existing ut Author: uncleGen <[email protected]> Closes apache#16052 from uncleGen/SPARK-18617.

…e. Receiver data should be deserialized properly ## What changes were proposed in this pull request? Fixed the potential SparkContext leak in `StreamingContextSuite.SPARK-18560 Receiver data should be deserialized properly` which was added in apache#16052. I also removed FakeByteArrayReceiver and used TestReceiver directly. ## How was this patch tested? Jenkins Author: Shixiong Zhu <[email protected]> Closes apache#16091 from zsxwing/SPARK-18617-follow-up.

SPARK-18617: Close "kryo auto pick" feature for Spark Streaming

d205ebd

uncleGen force-pushed the SPARK-18617 branch from e4416bd to d205ebd Compare November 29, 2016 06:47

rxin reviewed Nov 29, 2016

View reviewed changes

srowen reviewed Nov 29, 2016

View reviewed changes

uncleGen added 2 commits November 29, 2016 19:02

add comment and refactor ungainly code

491ba46

update comment

cd7e595

add unit test

39b4867

asfgit closed this in 56c82ed Nov 30, 2016

zsxwing reviewed Nov 30, 2016

View reviewed changes

zsxwing mentioned this pull request Dec 1, 2016

[SPARK-18617][Test]Fix flaky test: StreamingContextSuite.SPARK-18560 Receiver data should be deserialized properly #16091

Closed

uncleGen mentioned this pull request Dec 1, 2016

[SPARK-18617][BACKPORT] Follow up PR to Close "kryo auto pick" feature for Spark Streaming #16096

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-18617][CORE][STREAMING] Close "kryo auto pick" feature for Spark Streaming #16052

[SPARK-18617][CORE][STREAMING] Close "kryo auto pick" feature for Spark Streaming #16052

uncleGen commented Nov 29, 2016 •

edited

Loading

uncleGen commented Nov 29, 2016

SparkQA commented Nov 29, 2016

SparkQA commented Nov 29, 2016

rxin Nov 29, 2016

srowen Nov 29, 2016

uncleGen Nov 29, 2016

SparkQA commented Nov 29, 2016

zsxwing commented Nov 30, 2016

uncleGen commented Nov 30, 2016

SparkQA commented Nov 30, 2016

rxin commented Nov 30, 2016

rxin commented Nov 30, 2016

zsxwing Nov 30, 2016

uncleGen Nov 30, 2016 •

edited

Loading

uncleGen Nov 30, 2016

zsxwing Nov 30, 2016

uncleGen commented Nov 30, 2016

[SPARK-18617][CORE][STREAMING] Close "kryo auto pick" feature for Spark Streaming #16052

[SPARK-18617][CORE][STREAMING] Close "kryo auto pick" feature for Spark Streaming #16052

Conversation

uncleGen commented Nov 29, 2016 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

uncleGen commented Nov 29, 2016

SparkQA commented Nov 29, 2016

SparkQA commented Nov 29, 2016

rxin Nov 29, 2016

Choose a reason for hiding this comment

srowen Nov 29, 2016

Choose a reason for hiding this comment

uncleGen Nov 29, 2016

Choose a reason for hiding this comment

SparkQA commented Nov 29, 2016

zsxwing commented Nov 30, 2016

uncleGen commented Nov 30, 2016

SparkQA commented Nov 30, 2016

rxin commented Nov 30, 2016

rxin commented Nov 30, 2016

zsxwing Nov 30, 2016

Choose a reason for hiding this comment

uncleGen Nov 30, 2016 • edited Loading

Choose a reason for hiding this comment

uncleGen Nov 30, 2016

Choose a reason for hiding this comment

zsxwing Nov 30, 2016

Choose a reason for hiding this comment

uncleGen commented Nov 30, 2016

uncleGen commented Nov 29, 2016 •

edited

Loading

uncleGen Nov 30, 2016 •

edited

Loading