[SPARK-6602][Core]Replace Akka Serialization with Spark Serializer

[zsxwing]

Replace Akka Serialization with Spark Serializer and add unit tests.

[zsxwing]

/cc @rxin

[zsxwing]

This is the last PR for SPARK-6602

[SparkQA]

Test build #36265 has finished for PR 7159 at commit ecec410.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[SparkQA]

Test build #36264 has finished for PR 7159 at commit ff034d0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)
  • case class Cast(child: Expression, dataType: DataType) extends UnaryExpression with Logging

[rxin]

what is this thing used for?

[zsxwing]

org.apache.curator.test.TestingServer is from this artifact. An embedded ZooKeeper server for testing.

[adamv]

Should this be in test scope?

[zsxwing]

Yes. Good catch. I was thinking it but forgot to add it here.

[zsxwing]

Add this new method to RpcEnv for RpcEndpointRef deserialization

[SparkQA]

Test build #36398 has finished for PR 7159 at commit be3edb0.

• This patch fails MiMa tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[SparkQA]

Test build #36400 has finished for PR 7159 at commit 433115c.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[zsxwing]

retest this please

[rxin]

cc @andrewor14 can you review this? Thanks.

[SparkQA]

Test build #36408 has finished for PR 7159 at commit 433115c.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[SparkQA]

Test build #36502 has finished for PR 7159 at commit 9ef4af9.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[SparkQA]

Test build #36519 has finished for PR 7159 at commit 73251c6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[zsxwing]

ping @andrewor14

[andrewor14]

retest this please

[andrewor14]

can you explain why this is necessary?

[andrewor14]

I see, because now we no longer pass akka's Serialization, which has information about the actor system, into PersistenceEngine, so here we ensure that we're using the actor system's serializer.

But more generally, since we always serialize with JavaSerializer in the new code, why can't we always deserialize with the same thing? I just find it a little strange that we have to pass a closure into this method.

[zsxwing]

spark JavaSerializer is used to deserialize objects. However, it does not have an actor system in the current context. I need to use Akka JavaSerializer.currentSystem to put the current actor system into a thread-local variable.

[andrewor14]

but why do we need the actor system to deserialize it? Can't we just deserialize it with JavaSerializer? @rxin

[zsxwing]

but why do we need the actor system to deserialize it? Can't we just deserialize it with JavaSerializer?

Oh, that's because WorkerInfo and ApplicationInfo contain a reference to RpcEndpointRef.

[andrewor14]

I left one question, but LGTM otherwise.

[SparkQA]

Test build #37265 has finished for PR 7159 at commit 73251c6.

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

[JoshRosen]

/ping @jlewandowski, just want to give you a head's up about the binary-incompatible change to RecoveryModeFactory. I don't think that we can avoid this change, but I know that you're one of the few users / implementors of custom recovery modes and thought you'd want early notice.

[SparkQA]

Test build #37292 has finished for PR 7159 at commit fc0fca3.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • abstract class StandaloneRecoveryModeFactory(conf: SparkConf, serializer: Serializer)

[andrewor14]

Hi @zsxwing this LGTM. Feel free to merge it.

[rxin]

Thanks - I've merged it.

[jacek-lewandowski]

@JoshRosen this shouldn't be a problem for us. Thanks for pinging me anyway.