Updated to use scala queues insteadof google queues

Author: Ilya Ganelin

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -810,7 +810,7 @@ class DAGScheduler(
    *             Failure: String - The reason for the failure.
    *                                      
    */
-  def tryToSerializeRddDeps(rdd: RDD[_]): Array[RDDTrace] = {
+  private[spark] def tryToSerializeRddDeps(rdd: RDD[_]): Array[RDDTrace] = {
     SerializationHelper.tryToSerializeRddAndDeps(closureSerializer, rdd)
   }
 
@@ -869,20 +869,10 @@ class DAGScheduler(
     // where the JobConf/Configuration object is not thread-safe.
     var taskBinary: Broadcast[Array[Byte]] = null
 
-    // Check if RDD serialization debugging is enabled
-    val debugSerialization: Boolean = sc.getConf.getBoolean("spark.serializer.debug", false)
-    
     try {
       // For ShuffleMapTask, serialize and broadcast (rdd, shuffleDep). 
       // For ResultTask, serialize and broadcast (rdd, func).
 
-      // Before serialization print out the RDD and its references.
-      if (debugSerialization) {
-        SerializationHelper
-          .tryToSerializeRdd(closureSerializer, stage.rdd)
-          .fold(l => logDebug(l), r => {})
-      }
-       
       val taskBinaryBytes: Array[Byte] =
         if (stage.isShuffleMap) {
           closureSerializer.serialize((stage.rdd, stage.shuffleDep.get) : AnyRef).array()
@@ -893,10 +883,18 @@ class DAGScheduler(
     } catch {
       // In the case of a failure during serialization, abort the stage.
       case e: NotSerializableException =>
+        SerializationHelper
+          .tryToSerializeRdd(closureSerializer, stage.rdd)
+          .fold(l => logDebug(l), r => {})
+        
         abortStage(stage, "Task not serializable: " + e.toString)
         runningStages -= stage
         return
       case NonFatal(e) =>
+        SerializationHelper
+          .tryToSerializeRdd(closureSerializer, stage.rdd)
+          .fold(l => logDebug(l), r => {})
+        
         abortStage(stage, s"Task serialization failed: $e\n${e.getStackTraceString}")
         runningStages -= stage
         return

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

@@ -464,23 +464,18 @@ private[spark] class TaskSetManager(
             // We rely on the DAGScheduler to catch non-serializable closures and RDDs, so in here
             // we assume the task can be serialized without exceptions.
 
-            // Check if serialization debugging is enabled
-            val debugSerialization: Boolean = sched.sc.getConf.
-              getBoolean("spark.serializer.debug", false)
-
-            if (debugSerialization) {
+            Task.serializeWithDependencies(task, sched.sc.addedFiles, sched.sc.addedJars, ser)
+          } catch {
+            // If the task cannot be serialized, then there's no point to re-attempt the task,
+            // as it will always fail. So just abort the whole task-set and print a serialization
+            // trace to help identify the failure point.
+            case NonFatal(e) =>
               SerializationHelper.tryToSerialize(ser, task).fold (
                 l => logDebug("Un-serializable reference trace for " +
                   task.toString + ":\n" + l),
                 r => {}
               )
-            }
-
-            Task.serializeWithDependencies(task, sched.sc.addedFiles, sched.sc.addedJars, ser)
-          } catch {
-            // If the task cannot be serialized, then there's no point to re-attempt the task,
-            // as it will always fail. So just abort the whole task-set.
-            case NonFatal(e) =>
+              
               val msg = s"Failed to serialize task $taskId, not attempting to retry it."
               logError(msg, e)
               abort(s"$msg Exception during serialization: $e")

core/src/main/scala/org/apache/spark/util/ObjectWalker.scala

@@ -18,8 +18,6 @@ package org.apache.spark.util
 
 import java.lang.reflect.{Modifier, Field}
 
-import com.google.common.collect.Queues
-
 import scala.collection.mutable
 
 
@@ -30,7 +28,7 @@ import scala.collection.mutable
  * This code is based on code written by Josh Rosen found here:
  * https://gist.github.com/JoshRosen/d6a8972c99992e97d040
  */
-object ObjectWalker {
+private[spark] object ObjectWalker {
   def isTransient(field: Field): Boolean = Modifier.isTransient(field.getModifiers)
   def isStatic(field: Field): Boolean = Modifier.isStatic(field.getModifiers)
   def isPrimitive(field: Field): Boolean = field.getType.isPrimitive
@@ -50,13 +48,13 @@ object ObjectWalker {
    */
   def buildRefGraph(rootObj: AnyRef): mutable.LinkedList[AnyRef] = {
     val visitedRefs = mutable.Set[AnyRef]()
-    val toVisit = Queues.newArrayDeque[AnyRef]()
+    val toVisit = new mutable.Queue[AnyRef]()
     var results = mutable.LinkedList[AnyRef]()
 
-    toVisit.add(rootObj)
+    toVisit += rootObj
 
-    while (!toVisit.isEmpty) {
-      val obj : AnyRef = toVisit.pollFirst()
+    while (toVisit.nonEmpty) {
+      val obj : AnyRef = toVisit.dequeue()
       // Store the last parent reference to enable quick retrieval of the path to a broken node
 
       if (!visitedRefs.contains(obj)) {
@@ -66,26 +64,34 @@ object ObjectWalker {
         // Extract all the fields from the object that would be serialized. Transient and 
         // static references are not serialized and primitive variables will always be serializable
         // and will not contain further references.
-        
-        for (field <- getAllFields(obj.getClass)
-          .filterNot(isStatic)
-          .filterNot(isTransient)
-          .filterNot(isPrimitive)) {
+        for (field <- getFieldsToTest(obj)) {
           // Extract the field object and pass to the visitor
           val originalAccessibility = field.isAccessible
           field.setAccessible(true)
           val fieldObj = field.get(obj)
           field.setAccessible(originalAccessibility)
 
           if (fieldObj != null) {
-            toVisit.add(fieldObj)
+            toVisit += fieldObj
           }
         }  
       }
     }
     results
   }
 
+  /**
+   * Get the serialiazble fields from an object reference
+   * @param obj - Reference to the object fo rwhich to generate a serialization trace 
+   * @return a new Set containing the serializable fields of the object
+   */
+  def getFieldsToTest(obj: AnyRef): mutable.Set[Field] = {
+    getAllFields(obj.getClass)
+      .filterNot(isStatic)
+      .filterNot(isTransient)
+      .filterNot(isPrimitive)
+  }
+
   /**
    * Get all fields (including private ones) from this class and its superclasses.
    * @param cls - The class from which to retrieve fields

core/src/main/scala/org/apache/spark/util/RDDWalker.scala

@@ -17,8 +17,6 @@
 
 package org.apache.spark.util
 
-import com.google.common.collect.Queues
-
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.language.existentials
@@ -30,7 +28,7 @@ import org.apache.spark.rdd.RDD
  * accomplished by walking the object graph linking these RDDs. This is useful for debugging 
  * internal RDD references. See SPARK-3694.
  */
-object RDDWalker {
+private[spark] object RDDWalker {
   /**
    * Traverse the dependencies of the RDD and store them within an Array along with their depths.
    * Return this data structure and subsequently process it. 
@@ -40,20 +38,20 @@ object RDDWalker {
    */
   def walk(rddToWalk : RDD[_]): Array[(RDD[_], Int)] = {
 
-    val walkQueue = Queues.newArrayDeque[(RDD[_], Int)]()
+    val walkQueue = new mutable.Queue[(RDD[_], Int)]()
     val visited = mutable.Set[RDD[_]]()
 
     // Keep track of both the RDD and its depth in the traversal graph.
     val results = new ArrayBuffer[(RDD[_], Int)]()
     // Implement as a queue to perform a BFS  
-    walkQueue.addFirst(rddToWalk,0)
+    walkQueue += ((rddToWalk,0))
 
     while (!walkQueue.isEmpty) {
       // Pop from the queue 
-      val (rddToProcess : RDD[_], depth:Int) = walkQueue.pollFirst()
+      val (rddToProcess : RDD[_], depth:Int) = walkQueue.dequeue()
       if (!visited.contains(rddToProcess)) {
         visited.add(rddToProcess)
-        rddToProcess.dependencies.foreach(s => walkQueue.addFirst(s.rdd, depth + 1))
+        rddToProcess.dependencies.foreach(s => walkQueue += ((s.rdd, depth + 1)))
         results.append((rddToProcess, depth))
       }
     }

core/src/main/scala/org/apache/spark/util/SerializationHelper.scala

@@ -31,20 +31,20 @@ import org.apache.spark.serializer.SerializerInstance
 /**
  * This enumeration defines variables use to standardize debugging output
  */
-object SerializationState extends Enumeration {
+private[spark] object SerializationState extends Enumeration {
   type SerializationState = String
   val Failed = "Failed to serialize parent."
   val FailedDeps = "Failed to serialize dependencies."
   val Success = "Success"
 }
 
-case class RDDTrace (rdd : RDD[_], depth : Int, result : SerializationHelper.SerializedRef)
+private[spark] case class RDDTrace (rdd : RDD[_], depth : Int, result : SerializationHelper.SerializedRef)
 
 /**
  * This class is designed to encapsulate some utilities to facilitate debugging serialization 
  * problems in the DAGScheduler and the TaskSetManager. See SPARK-3694.
  */
-object SerializationHelper {
+private[spark] object SerializationHelper {
   type PathToRef = mutable.LinkedList[AnyRef]
   type BrokenRef = (AnyRef, PathToRef)
   type SerializedRef = Either[String, ByteBuffer]
@@ -249,7 +249,7 @@ object SerializationHelper {
 
   def refString(ref: AnyRef): String = {
     val refCode = System.identityHashCode(ref)
-    "Ref (" + ref.toString + ", Hash: " + refCode + ")"
+    "Ref (" + ref.toString + ")"
   }
 
   /**