ConcurrentHashMap -> SynchronizedMap

Author: JoshRosen

streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala

@@ -21,7 +21,6 @@ import java.io.{IOException, ObjectInputStream}
 import java.util.concurrent.ConcurrentHashMap
 
 import scala.collection.mutable
-import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 
 import org.apache.hadoop.fs.{FileSystem, Path, PathFilter}
@@ -98,7 +97,7 @@ class FileInputDStream[K: ClassTag, V: ClassTag, F <: NewInputFormat[K,V] : Clas
   // Map of batch-time to selected file info for the remembered batches
   // This is a concurrent map because it's also accessed in unit tests
   @transient private[streaming] var batchTimeToSelectedFiles =
-    new ConcurrentHashMap[Time, Array[String]].asScala
+    new mutable.HashMap[Time, Array[String]] with mutable.SynchronizedMap[Time, Array[String]]
 
   // Set of files that were selected in the remembered batches
   @transient private var recentlySelectedFiles = new mutable.HashSet[String]()
@@ -273,7 +272,8 @@ class FileInputDStream[K: ClassTag, V: ClassTag, F <: NewInputFormat[K,V] : Clas
     logDebug(this.getClass().getSimpleName + ".readObject used")
     ois.defaultReadObject()
     generatedRDDs = new mutable.HashMap[Time, RDD[(K,V)]] ()
-    batchTimeToSelectedFiles = new ConcurrentHashMap[Time, Array[String]].asScala
+    batchTimeToSelectedFiles =
+      new mutable.HashMap[Time, Array[String]] with mutable.SynchronizedMap[Time, Array[String]]
     recentlySelectedFiles = new mutable.HashSet[String]()
     fileToModTime = new TimeStampedHashMap[String, Long](true)
   }