[SPARK-18560][CORE][STREAMING][WIP] Receiver data can not be deserialized properly.

[uncleGen]

What changes were proposed in this pull request?

My spark streaming job can run correctly on Spark 1.6.1, but it can not run properly on Spark master branch, with following exception:

16/11/22 19:20:15 ERROR executor.Executor: Exception in task 4.3 in stage 6.0 (TID 87)
com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 13994
	at com.esotericsoftware.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:137)
	at com.esotericsoftware.kryo.Kryo.readClass(Kryo.java:670)
	at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:781)
	at org.apache.spark.serializer.KryoDeserializationStream.readObject(KryoSerializer.scala:243)
	at org.apache.spark.serializer.DeserializationStream$$anon$1.getNext(Serializer.scala:169)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at org.apache.spark.util.Utils$.getIteratorSize(Utils.scala:1760)
	at org.apache.spark.rdd.RDD$$anonfun$count$1.apply(RDD.scala:1150)
	at org.apache.spark.rdd.RDD$$anonfun$count$1.apply(RDD.scala:1150)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1943)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1943)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
	at org.apache.spark.scheduler.Task.run(Task.scala:108)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)

Dig deep into relevant implementation, I find the type of data received by Receiver is erased. And in Spark2.x, framework can choose a appropriate Serializer from JavaSerializer and KryoSerializer base on the type of data.

At the Receiver side, the type of data is erased to be Object, so framework will choose JavaSerializer, with following code:

def canUseKryo(ct: ClassTag[_]): Boolean = {
    primitiveAndPrimitiveArrayClassTags.contains(ct) || ct == stringClassTag
  }

  def getSerializer(ct: ClassTag[_]): Serializer = {
    if (canUseKryo(ct)) {
      kryoSerializer
    } else {
      defaultSerializer
    }
  }

At task side, we can get correct data type, and framework will choose KryoSerializer if possible, with following supported type:

private[this] val stringClassTag: ClassTag[String] = implicitly[ClassTag[String]]
private[this] val primitiveAndPrimitiveArrayClassTags: Set[ClassTag[_]] = {
    val primitiveClassTags = Set[ClassTag[_]](
      ClassTag.Boolean,
      ClassTag.Byte,
      ClassTag.Char,
      ClassTag.Double,
      ClassTag.Float,
      ClassTag.Int,
      ClassTag.Long,
      ClassTag.Null,
      ClassTag.Short
    )
    val arrayClassTags = primitiveClassTags.map(_.wrap)
    primitiveClassTags ++ arrayClassTags
  }

In my case, the type of data is Byte Array.

This problem stems from SPARK-13990, a patch to have Spark automatically pick the "best" serializer when caching RDDs.

How was this patch tested?

update existing unit test

[uncleGen]

cc @JoshRosen

[SparkQA]

Test build #69063 has finished for PR 15992 at commit cbc4fc6.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• abstract class Receiver[T](val storageLevel: StorageLevel)(implicit t: ClassTag[T])

[SparkQA]

Test build #69064 has finished for PR 15992 at commit 489766e.

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

[SparkQA]

Test build #69066 has finished for PR 15992 at commit 9e89e16.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #69067 has finished for PR 15992 at commit ec36a0e.

• This patch fails MiMa tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• abstract class Receiver[T](val storageLevel: StorageLevel)(implicit t: ClassTag[T])

[SparkQA]

Test build #69070 has finished for PR 15992 at commit 6f64afd.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #69072 has finished for PR 15992 at commit 037f10d.

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

[uncleGen]

cc @zsxwing and @JoshRosen

[jerryshao]

I'm wondering will this change the signature of Receiver and break the existing code by user?

[uncleGen]

As far as I know, this change will indeed broke custom receiver implementation in Java, just like the update in JavaCustomReceiver.java. Besides, in my humble opinion, I did not encounter any problem in my local test.

[jerryshao]

I see, so in that case we should carefully think about this change, at least keep compatible with existing code.

[uncleGen]

It is really trickier to add ClassTag in many place to support automatically-pick-best-serializer feature.

[jerryshao]

I'm thinking if we could treat Spark Streaming as a special case to turn off type based auto serializer choosing mechanism. Current fix introduces lots of changes in the streaming code and may potentially break user's existing code. Just my two cents.

[uncleGen]

@jerryshao yep, the simplest way is to turn it off. Generally, this is a followup patch for #11755. I tends to support automatically-pick-best-serializer feature here to get better performance, if there is no big damage to existing code, sooner rather than later.

[uncleGen]

@JoshRosen and @zsxwing waiting for your feedback.

[uncleGen]

NO one will review this PR? @srowen Could you please call someone to review this? Thanks!

[srowen]

I am not familiar enough with this code to review it. I do think @JoshRosen is the right person given https://issues.apache.org/jira/browse/SPARK-13990 and believe he's said he will start reviewing again this week after the holiday

[zsxwing]

Sorry for the delay.

My high level comment is this is a critical bug and we also need to fix 2.0.x. However, we usually don't want to break APIs between maintenance releases. So I'm thinking the first step we should do is adding a configuration at https://github.com/apache/spark/pull/11755/files#diff-82b3c6a9530bd021cfb3d3f15056be41R49 and disable it for Streaming.

[uncleGen]

@zsxwing Maybe, it is indeed a well-advised solution. I will provide another PR first to add a configuration for streaming. IMHO, I will keep this PR WIP for farther discussion and optimization. What is your opinion?

[zsxwing]

I think this PR has one issue. There are two places requiring ClassTag. The first one is creating Receiver, the other one is calling receiverStream. If they are inconsistent (e.g., writing some Java codes to create Receiver, and call receiverStream using Scala codes), then it will crash.

[zsxwing]

@uncleGen could you close this one, please? The current workaround in #15992 looks enough. You can create a new one if you have a better solution for Java APIs.