[HADP-56155] Support scala 2.13 (apache#670)

Author: wangyum

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

@@ -276,7 +276,7 @@ abstract private[spark] class ContextCleanupWorker(sc: SparkContext, name: Strin
   @volatile private var stopped = false
 
   private val cleaningThread = new Thread() {
-    override def run() {
+    override def run(): Unit = {
       keepCleaning()
     }
   }

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

@@ -276,7 +276,7 @@ private class ShuffleStatus(
    * Removes all shuffle outputs associated with this host. Note that this will also remove
    * outputs which are served by an external shuffle server (if one exists).
    */
-  def removeOutputsOnHost(host: String): Seq[Int] = withWriteLock {
+  def removeOutputsOnHost(host: String): ArrayBuffer[Int] = withWriteLock {
     logDebug(s"Removing outputs for host ${host}")
     val removedMapIds = removeOutputsByFilter(x => x.host == host)
     removeMergeResultsByFilter(x => x.host == host)
@@ -288,7 +288,7 @@ private class ShuffleStatus(
    * remove outputs which are served by an external shuffle server (if one exists), as they are
    * still registered with that execId.
    */
-  def removeOutputsOnExecutor(execId: String): Seq[Int] = withWriteLock {
+  def removeOutputsOnExecutor(execId: String): ArrayBuffer[Int] = withWriteLock {
     logDebug(s"Removing outputs for execId ${execId}")
     removeOutputsByFilter(x => x.executorId == execId)
   }
@@ -297,7 +297,7 @@ private class ShuffleStatus(
    * Removes all shuffle outputs which satisfies the filter. Note that this will also
    * remove outputs which are served by an external shuffle server (if one exists).
    */
-  def removeOutputsByFilter(f: BlockManagerId => Boolean): Seq[Int] = withWriteLock {
+  def removeOutputsByFilter(f: BlockManagerId => Boolean): ArrayBuffer[Int] = withWriteLock {
     val removedMapIds = ArrayBuffer[Int]()
     for (mapIndex <- mapStatuses.indices) {
       val currentMapStatus = mapStatuses(mapIndex)

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

@@ -2462,7 +2462,7 @@ class SparkContext(config: SparkConf) extends Logging {
     logInfo("Successfully stopped SparkContext")
   }
 
-  def logStop(name: String)(stop: => Unit) {
+  def logStop(name: String)(stop: => Unit): Unit = {
     logInfo(s"Stopping $name")
     Utils.tryLogNonFatalError {
       stop

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

@@ -292,5 +292,5 @@ private[spark] class TaskContextImpl(
 
   private[spark] override def fetchFailed: Option[FetchFailedException] = _fetchFailedException
 
-  private[spark] override def getLocalProperties(): Properties = localProperties
+  private[spark] override def getLocalProperties: Properties = localProperties
 }

core/src/main/scala/org/apache/spark/deploy/history/ViewPointFsHistoryProvider.scala

@@ -219,7 +219,7 @@ private[history] class ViewPointFsHistoryProvider(conf: SparkConf, clock: Clock)
                     s"$logName, lastScanSize: $lastScanSize, currentSize: ${fileStatus.getLen()}")
                   val task: Future[Unit] = replayExecutor.submit(new Runnable {
                     override def run(): Unit = updateEventLog(fileStatus)
-                  }, Unit)
+                  }, ())
                   tasks = tasks :+ task
                   pendingReplayTasksCount.addAndGet(1)
                 }
@@ -353,7 +353,7 @@ private[history] class ViewPointFsHistoryProvider(conf: SparkConf, clock: Clock)
           try {
             val task: Future[Unit] = replayExecutor.submit(new Runnable {
               override def run(): Unit = mergeApplicationListing(entry, newLastScanTime)
-            }, Unit)
+            }, ())
             Some(task -> entry.getPath)
           } catch {
             // let the iteration over the updated entries break, since an exception on
@@ -653,7 +653,7 @@ private[history] class ViewPointFsHistoryProvider(conf: SparkConf, clock: Clock)
           logInfo(s"Removed ${appId} from favorite list")
         }
       }
-    }, Unit)
+    }, ())
   }
 
   override def getFavoriteSessions(

core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala

@@ -563,7 +563,7 @@ object HadoopMapReduceCommitProtocol extends Logging {
         logInfo(s"Merging data from $from to $to")
         rename(fs, from, to)
       }
-    }, Unit)
+    }, ())
   }
 
   @throws[IOException]

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

@@ -97,7 +97,7 @@ class OrderedRDDFunctions[K : Ordering : ClassTag,
     def inRange(k: K): Boolean = ordering.gteq(k, lower) && ordering.lteq(k, upper)
 
     val rddToFilter: RDD[P] = self.partitioner match {
-      case Some(rp: RangePartitioner[K, V]) =>
+      case Some(rp: RangePartitioner[_, _]) =>
         val partitionIndices = (rp.getPartition(lower), rp.getPartition(upper)) match {
           case (l, u) => Math.min(l, u) to Math.max(l, u)
         }

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

@@ -79,7 +79,7 @@ private[spark] class AnalyticsTaskSchedulerImpl(
   private val executorStatusLock = new ReentrantReadWriteLock()
 
   override def runningTasksByExecutors: Map[String, Int] = {
-    executorIdToRunningTaskIds.toMap.mapValues(_.size)
+    executorIdToRunningTaskIds.toMap.mapValues(_.size).toMap
   }
 
   // The set of executors we have on each host; this is used to compute hostsAlive, which
@@ -552,7 +552,7 @@ private[spark] class AnalyticsTaskSchedulerImpl(
     if (tasks.nonEmpty) {
       hasLaunchedTask = true
     }
-    tasks
+    tasks.map(_.toSeq)
   } finally {
     executorStatusLock.readLock().unlock()
   }
@@ -930,7 +930,7 @@ private[spark] class AnalyticsTaskSchedulerImpl(
     abortTimer.cancel()
   }
 
-  def checkTaskSetHealth() {
+  def checkTaskSetHealth(): Unit = {
     val taskSets = rootPool.getTaskSetQueue
     for (taskSet <- taskSets) {
       if (taskSet.isZombie) {

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

@@ -21,7 +21,7 @@ import java.io.NotSerializableException
 import java.nio.ByteBuffer
 import java.util.concurrent.ConcurrentLinkedQueue
 
-import scala.collection.mutable.{ArrayBuffer, HashMap}
+import scala.collection.mutable.{ArrayBuffer, Buffer, HashMap}
 import scala.math.max
 import scala.util.control.NonFatal
 
@@ -162,7 +162,7 @@ private[spark] class AnalyticsTaskSetManager(
       index: Int,
       taskLocality: TaskLocality.Value,
       speculative: Boolean,
-      availableResources: Map[String, Seq[String]] = Map.empty): TaskDescription = {
+      availableResources: Map[String, Buffer[String]] = Map.empty): TaskDescription = {
     // Found a task; do some bookkeeping and return a task description
     val task = tasks(index)
     val taskId = sched.newTaskId()
@@ -443,7 +443,7 @@ private[spark] class AnalyticsTaskSetManager(
     }
   }
 
-  override def getResSufficientSortedTaskSetQueue: ArrayBuffer[TaskSetManager] = {
+  override def getResSufficientSortedTaskSetQueue(): ArrayBuffer[TaskSetManager] = {
     if (hasMoreComputeResources) {
       val sortedTaskSetQueue = new ArrayBuffer[TaskSetManager]()
       sortedTaskSetQueue += this

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

@@ -402,7 +402,7 @@ private[spark] class DAGScheduler(
           lightAccumUpdates += accum
         }
       )
-      (heavyAccumUpdates, lightAccumUpdates, accumUpdatesForDAG)
+      (heavyAccumUpdates.toSeq, lightAccumUpdates.toSeq, accumUpdatesForDAG.toSeq)
     }
 
     def taskMetricsFromAccumulators(accumUpdates: Seq[AccumulatorV2[_, _]]): TaskMetrics = {