Using the scala collections

compiler was not able to differentiate between the scala and java api

core/src/main/scala/org/apache/spark/api/java/JavaUtils.scala

@@ -31,14 +31,15 @@ private[spark] object JavaUtils {
     }
 
   // Workaround for SPARK-3926 / SI-8911
-  def mapAsSerializableJavaMap[A, B](underlying: collection.Map[A, B]): SerializableMapWrapper[A, B]
-    = new SerializableMapWrapper(underlying)
+  def mapAsSerializableJavaMap[A, B](underlying: scala.collection.Map[A, B]):
+  SerializableMapWrapper[A, B]
+  = new SerializableMapWrapper(underlying)
 
   // Implementation is copied from scala.collection.convert.Wrappers.MapWrapper,
   // but implements java.io.Serializable. It can't just be subclassed to make it
   // Serializable since the MapWrapper class has no no-arg constructor. This class
   // doesn't need a no-arg constructor though.
-  class SerializableMapWrapper[A, B](underlying: collection.Map[A, B])
+  class SerializableMapWrapper[A, B](underlying: scala.collection.Map[A, B])
     extends ju.AbstractMap[A, B] with java.io.Serializable { self =>
 
     override def size: Int = underlying.size

core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala

@@ -461,7 +461,7 @@ private[spark] object PythonRDD extends Logging {
   JavaRDD[Array[Byte]] = {
     val file = new DataInputStream(new FileInputStream(filename))
     try {
-      val objs = new collection.mutable.ArrayBuffer[Array[Byte]]
+      val objs = new scala.collection.mutable.ArrayBuffer[Array[Byte]]
       try {
         while (true) {
           val length = file.readInt()

core/src/main/scala/org/apache/spark/deploy/mesos/ui/DriverPage.scala

@@ -114,7 +114,7 @@ private[ui] class DriverPage(parent: MesosClusterUI) extends WebUIPage("driver")
     }.getOrElse(Seq[Node]())
   }
 
-  private def propertiesRow(properties: collection.Map[String, String]): Seq[Node] = {
+  private def propertiesRow(properties: scala.collection.Map[String, String]): Seq[Node] = {
     properties.map { case (k, v) =>
       <tr>
         <td>{k}</td><td>{v}</td>

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.scheduler
 
-import collection.mutable.ArrayBuffer
+import scala.collection.mutable.ArrayBuffer
 
 import org.apache.spark.annotation.DeveloperApi
 

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

@@ -509,7 +509,7 @@ private[spark] class BlockManagerInfo(
   def blocks: JHashMap[BlockId, BlockStatus] = _blocks
 
   // This does not include broadcast blocks.
-  def cachedBlocks: collection.Set[BlockId] = _cachedBlocks
+  def cachedBlocks: scala.collection.Set[BlockId] = _cachedBlocks
 
   override def toString: String = "BlockManagerInfo " + timeMs + " " + _remainingMem