Import, comments, and style fixes (minor)

Author: andrewor14

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

@@ -112,8 +112,8 @@ private[spark] abstract class MapOutputTracker(conf: SparkConf) extends Logging
   }
 
   /**
-   * Called from executors to get the server URIs and
-   * output sizes of the map outputs of a given shuffle
+   * Called from executors to get the server URIs and output sizes of the map outputs of
+   * a given shuffle.
    */
   def getServerStatuses(shuffleId: Int, reduceId: Int): Array[(BlockManagerId, Long)] = {
     val statuses = mapStatuses.get(shuffleId).orNull
@@ -218,10 +218,9 @@ private[spark] class MapOutputTrackerMaster(conf: SparkConf)
   private var cacheEpoch = epoch
 
   /**
-   * Timestamp based HashMap for storing mapStatuses and cached serialized statuses
-   * in the master, so that statuses are dropped only by explicit deregistering or
-   * by TTL-based cleaning (if set). Other than these two
-   * scenarios, nothing should be dropped from this HashMap.
+   * Timestamp based HashMap for storing mapStatuses and cached serialized statuses in the master,
+   * so that statuses are dropped only by explicit deregistering or by TTL-based cleaning (if set).
+   * Other than these two scenarios, nothing should be dropped from this HashMap.
    */
   protected val mapStatuses = new TimeStampedHashMap[Int, Array[MapStatus]]()
   private val cachedSerializedStatuses = new TimeStampedHashMap[Int, Array[Byte]]()

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -230,6 +230,7 @@ class SparkContext(
 
   private[spark] val cleaner = new ContextCleaner(this)
   cleaner.start()
+
   postEnvironmentUpdate()
 
   /** A default Hadoop Configuration for the Hadoop code (e.g. file systems) that we reuse. */
@@ -773,7 +774,7 @@ class SparkContext(
    * filesystems), an HTTP, HTTPS or FTP URI, or local:/path for a file on every worker node.
    */
   def addJar(path: String) {
-    if (path  == null) {
+    if (path == null) {
       logWarning("null specified as parameter to addJar")
     } else {
       var key = ""

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -185,6 +185,7 @@ object SparkEnv extends Logging {
     } else {
       new MapOutputTrackerWorker(conf)
     }
+
     // Have to assign trackerActor after initialization as MapOutputTrackerActor
     // requires the MapOutputTracker itself
     mapOutputTracker.trackerActor = registerOrLookup(

core/src/main/scala/org/apache/spark/broadcast/BroadcastFactory.scala

@@ -27,8 +27,8 @@ import org.apache.spark.SparkConf
  * entire Spark job.
  */
 trait BroadcastFactory {
-  def initialize(isDriver: Boolean, conf: SparkConf, securityMgr: SecurityManager): Unit
+  def initialize(isDriver: Boolean, conf: SparkConf, securityMgr: SecurityManager)
   def newBroadcast[T](value: T, isLocal: Boolean, id: Long): Broadcast[T]
   def unbroadcast(id: Long, removeFromDriver: Boolean)
-  def stop(): Unit
+  def stop()
 }

core/src/main/scala/org/apache/spark/broadcast/HttpBroadcast.scala

@@ -90,7 +90,7 @@ private[spark] object HttpBroadcast extends Logging {
   private var securityManager: SecurityManager = null
 
   // TODO: This shouldn't be a global variable so that multiple SparkContexts can coexist
-  val files = new TimeStampedHashSet[String]
+  private val files = new TimeStampedHashSet[String]
   private var cleaner: MetadataCleaner = null
 
   private val httpReadTimeout = TimeUnit.MILLISECONDS.convert(5, TimeUnit.MINUTES).toInt
@@ -195,7 +195,7 @@ private[spark] object HttpBroadcast extends Logging {
   def unpersist(id: Long, removeFromDriver: Boolean) = synchronized {
     SparkEnv.get.blockManager.master.removeBroadcast(id, removeFromDriver)
     if (removeFromDriver) {
-      val file = new File(broadcastDir, BroadcastBlockId(id).name)
+      val file = getFile(id)
       files.remove(file.toString)
       deleteBroadcastFile(file)
     }
@@ -217,10 +217,9 @@ private[spark] object HttpBroadcast extends Logging {
     }
   }
 
-  /** Delete the given broadcast file. */
   private def deleteBroadcastFile(file: File) {
     try {
-      if (!file.exists()) {
+      if (!file.exists) {
         logWarning("Broadcast file to be deleted does not exist: %s".format(file))
       } else if (file.delete()) {
         logInfo("Deleted broadcast file: %s".format(file))
@@ -229,7 +228,7 @@ private[spark] object HttpBroadcast extends Logging {
       }
     } catch {
       case e: Exception =>
-        logWarning("Exception while deleting broadcast file: %s".format(file), e)
+        logError("Exception while deleting broadcast file: %s".format(file), e)
     }
   }
 

core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

@@ -72,7 +72,7 @@ private[spark] class TorrentBroadcast[T](@transient var value_ : T, isLocal: Boo
   }
 
   /**
-   * Remove all persisted state associated with this HTTP broadcast.
+   * Remove all persisted state associated with this Torrent broadcast.
    * @param removeFromDriver Whether to remove state from the driver.
    */
   override def unpersist(removeFromDriver: Boolean) {
@@ -177,13 +177,12 @@ private[spark] class TorrentBroadcast[T](@transient var value_ : T, isLocal: Boo
 }
 
 private[spark] object TorrentBroadcast extends Logging {
+  private lazy val BLOCK_SIZE = conf.getInt("spark.broadcast.blockSize", 4096) * 1024
   private var initialized = false
   private var conf: SparkConf = null
 
-  lazy val BLOCK_SIZE = conf.getInt("spark.broadcast.blockSize", 4096) * 1024
-
   def initialize(_isDriver: Boolean, conf: SparkConf) {
-    TorrentBroadcast.conf = conf //TODO: we might have to fix it in tests
+    TorrentBroadcast.conf = conf // TODO: we might have to fix it in tests
     synchronized {
       if (!initialized) {
         initialized = true

core/src/main/scala/org/apache/spark/rdd/RDD.scala

@@ -158,7 +158,7 @@ abstract class RDD[T: ClassTag](
    */
   def unpersist(blocking: Boolean = true): RDD[T] = {
     logInfo("Removing RDD " + id + " from persistence list")
-    sc.unpersistRDD(this.id, blocking)
+    sc.unpersistRDD(id, blocking)
     storageLevel = StorageLevel.NONE
     this
   }
@@ -1128,4 +1128,5 @@ abstract class RDD[T: ClassTag](
   def toJavaRDD() : JavaRDD[T] = {
     new JavaRDD(this)(elementClassTag)
   }
+
 }

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

@@ -1090,7 +1090,6 @@ class DAGScheduler(
       eventProcessActor ! StopDAGScheduler
     }
     taskScheduler.stop()
-    listenerBus.stop()
   }
 }
 

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

@@ -49,8 +49,8 @@ private[spark] class BlockManager(
     maxMemory: Long,
     val conf: SparkConf,
     securityManager: SecurityManager,
-    mapOutputTracker: MapOutputTracker
-  ) extends Logging {
+    mapOutputTracker: MapOutputTracker)
+  extends Logging {
 
   val shuffleBlockManager = new ShuffleBlockManager(this)
   val diskBlockManager = new DiskBlockManager(shuffleBlockManager,

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

@@ -82,7 +82,7 @@ class BlockManagerMaster(var driverActor: ActorRef, conf: SparkConf) extends Log
 
   /**
    * Check if block manager master has a block. Note that this can be used to check for only
-   * those blocks that are expected to be reported to block manager master.
+   * those blocks that are reported to block manager master.
    */
   def contains(blockId: BlockId) = {
     !getLocations(blockId).isEmpty