Mon Aug 22 13:54:26 PDT 2016

Author: ericl

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -977,8 +977,7 @@ private[spark] class BlockManager(
           // [SPARK-16550] Erase the typed classTag when using default serialization, since
           // NettyBlockRpcServer crashes when deserializing repl-defined classes.
           // TODO(ekl) remove this once the classloader issue on the remote end is fixed.
-          val remoteClassTag = classTag
-          if (!serializerManager.canUseKryo(classTag)) {
+          val remoteClassTag = if (!serializerManager.canUseKryo(classTag)) {
             scala.reflect.classTag[Any]
           } else {
             classTag

core/src/test/scala/org/apache/spark/DistributedSuite.scala

@@ -151,6 +151,7 @@ class DistributedSuite extends SparkFunSuite with Matchers with LocalSparkContex
 
   private def testCaching(storageLevel: StorageLevel): Unit = {
     sc = new SparkContext(clusterUrl, "test")
+    sc.jobProgressListener.waitUntilExecutorsUp(2, 30000)
     val data = sc.parallelize(1 to 1000, 10)
     val cachedData = data.persist(storageLevel)
     assert(cachedData.count === 1000)