refactor(jobserver): Remove redundant caching code (spark-jobserver#949)

Mesos cluster change[1] introduced this redundant
caching and can be safely removed. Now, FileCacher
is the only class responsible for caching jars. The
discussion regarding the consent of stakeholders is
here [2].

Also removing all the instances of GetBinaryContent
because it was only used by redundant change.

[1] https://github.com/spark-jobserver/spark-jobserver/pull/681/files
[2] spark-jobserver@9084053#commitcomment-25033986

Author: bsikander

job-server/src/main/scala/spark/jobserver/JobCache.scala

@@ -1,8 +1,7 @@
 package spark.jobserver
 
-import java.io.{File, IOException}
+import java.io.File
 import java.net.URL
-import java.nio.file.{Files, Paths}
 
 import akka.actor.ActorRef
 import akka.util.Timeout
@@ -48,25 +47,7 @@ class JobCacheImpl(maxEntries: Int,
       val jarPathReq =
         (dao ? JobDAOActor.GetBinaryPath(appName, BinaryType.Jar, uploadTime)).mapTo[JobDAOActor.BinaryPath]
       val jarPath = Await.result(jarPathReq, daoAskTimeout.duration).binPath
-      logger.info("End of get jar path for app {}, uploadTime {}, jarPath {}", appName, uploadTime, jarPath)
-      val jarFile = Paths.get(jarPath)
-      if (!Files.exists(jarFile)) {
-        logger.info("Local jar path {} not exist, fetch binary content from remote actor", jarPath)
-        val jarBinaryReq = (dao ? JobDAOActor.GetBinaryContent(appName, BinaryType.Jar, uploadTime))
-          .mapTo[JobDAOActor.BinaryContent]
-        val binaryJar = Await.result(jarBinaryReq, daoAskTimeout.duration)
-        logger.info("Writing {} bytes to file {}", binaryJar.content.size, jarFile.toAbsolutePath.toString)
-        try {
-          if (!Files.exists(jarFile.getParent)) {
-            logger.info("Creating cache dir {}", jarFile.getParent.toAbsolutePath.toString)
-            Files.createDirectories(jarFile.getParent)
-          }
-          Files.write(jarFile, binaryJar.content)
-        } catch {
-          case e: IOException => logger.error("Write to path {} error {}", jarPath: Any, e)
-        }
-      }
-      val jarFilePath = jarFile.toAbsolutePath.toString
+      val jarFilePath = new File(jarPath).getAbsolutePath()
       sparkContext.addJar(jarFilePath) // Adds jar for remote executors
       loader.addURL(new URL("file:" + jarFilePath)) // Now jar added for local loader
       val constructor = JarUtils.loadClassOrObject[spark.jobserver.api.SparkJobBase](classPath, loader)

job-server/src/main/scala/spark/jobserver/io/JobCassandraDAO.scala

@@ -3,7 +3,6 @@ package spark.jobserver.io
 import java.io.File
 import java.net.InetSocketAddress
 import java.nio.ByteBuffer
-import java.nio.file.{Files, Paths}
 import java.util.UUID
 
 import scala.collection.convert.WrapAsJava
@@ -410,16 +409,4 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
 
     session.execute(runningJobsView)
   }
-
-  override def getBinaryContent(appName: String, binaryType: BinaryType,
-                                uploadTime: DateTime): Array[Byte] = {
-    val jarFile = new File(rootDir, createBinaryName(appName, binaryType, uploadTime))
-    if (!jarFile.exists()) {
-      val binBytes = fetchBinary(appName, binaryType, uploadTime)
-      cacheBinary(appName, binaryType, uploadTime, binBytes)
-      binBytes
-    } else {
-      Files.readAllBytes(Paths.get(jarFile.getAbsolutePath))
-    }
-  }
 }

job-server/src/main/scala/spark/jobserver/io/JobDAO.scala

@@ -184,14 +184,4 @@ trait JobDAO {
    * @return Some(lastUploadedTime) if the app exists and the list of times is nonempty, None otherwise
    */
   def getLastUploadTimeAndType(appName: String): Option[(DateTime, BinaryType)]
-
-  /**
-    * Fetch submited jar or egg content for remote driver and JobManagerActor to cache in local
-    * @param appName
-    * @param uploadTime
-    * @return
-    */
-  def getBinaryContent(appName: String,
-                       binaryType: BinaryType,
-                       uploadTime: DateTime): Array[Byte]
 }

job-server/src/main/scala/spark/jobserver/io/JobDAOActor.scala

@@ -28,9 +28,6 @@ object JobDAOActor {
   case class GetBinaryPath(appName: String,
                            binaryType: BinaryType,
                            uploadTime: DateTime) extends JobDAORequest
-  case class GetBinaryContent(appName: String,
-                              binaryType: BinaryType,
-                              uploadTime: DateTime) extends JobDAORequest
 
   case class SaveJobInfo(jobInfo: JobInfo) extends JobDAORequest
   case class GetJobInfos(limit: Int) extends JobDAORequest
@@ -45,7 +42,6 @@ object JobDAOActor {
   sealed trait JobDAOResponse
   case class Apps(apps: Map[String, (BinaryType, DateTime)]) extends JobDAOResponse
   case class BinaryPath(binPath: String) extends JobDAOResponse
-  case class BinaryContent(content: Array[Byte]) extends JobDAOResponse
   case class JobInfos(jobInfos: Seq[JobInfo]) extends JobDAOResponse
   case class JobConfig(jobConfig: Option[Config]) extends JobDAOResponse
   case class LastUploadTimeAndType(uploadTimeAndType: Option[(DateTime, BinaryType)]) extends JobDAOResponse
@@ -92,9 +88,6 @@ class JobDAOActor(dao: JobDAO) extends InstrumentedActor {
     case GetLastUploadTimeAndType(appName) =>
       sender() ! LastUploadTimeAndType(dao.getLastUploadTimeAndType(appName))
 
-    case GetBinaryContent(appName, binaryType, uploadTime) =>
-      sender() ! BinaryContent(dao.getBinaryContent(appName, binaryType, uploadTime))
-
     case CleanContextJobInfos(contextName, endTime) =>
       dao.cleanRunningJobInfosForContext(contextName, endTime)
   }

job-server/src/main/scala/spark/jobserver/io/JobFileDAO.scala

@@ -1,7 +1,6 @@
 package spark.jobserver.io
 
 import java.io._
-import java.nio.file.{Files, Paths}
 
 import com.typesafe.config._
 import org.joda.time.DateTime
@@ -238,11 +237,6 @@ class JobFileDAO(config: Config) extends JobDAO {
     ConfigFactory.parseString(in.readUTF)
   )
 
-  override def getBinaryContent(appName: String, binaryType: BinaryType,
-                                uploadTime: DateTime): Array[Byte] = {
-    Files.readAllBytes(Paths.get(retrieveBinaryFile(appName, binaryType, uploadTime)))
-  }
-
   /**
     * Delete a jar.
     *

job-server/src/main/scala/spark/jobserver/io/JobSqlDAO.scala

@@ -1,7 +1,6 @@
 package spark.jobserver.io
 
 import java.io.File
-import java.nio.file.{Files, Paths}
 import java.sql.{Blob, Timestamp}
 import javax.sql.DataSource
 import javax.sql.rowset.serial.SerialBlob
@@ -374,23 +373,4 @@ class JobSqlDAO(config: Config) extends JobDAO with FileCacher {
       r.map(jobInfoFromRow).headOption
     }
   }
-
-  /**
-    * Fetch submited jar or egg content for remote driver and JobManagerActor to cache in local
-    *
-    * @param appName
-    * @param uploadTime
-    * @return
-    */
-  override def getBinaryContent(appName: String, binaryType: BinaryType,
-                                uploadTime: DateTime): Array[Byte] = {
-    val jarFile = new File(rootDir, createBinaryName(appName, binaryType, uploadTime))
-    if (!jarFile.exists()) {
-      val binBytes = Await.result(fetchBinary(appName, binaryType, uploadTime), 60.seconds)
-      cacheBinary(appName, binaryType, uploadTime, binBytes)
-      binBytes
-    } else {
-      Files.readAllBytes(Paths.get(jarFile.getAbsolutePath))
-    }
-  }
 }

job-server/src/test/scala/spark/jobserver/InMemoryDAO.scala

@@ -86,12 +86,7 @@ class InMemoryDAO extends JobDAO {
     Await.result(getApps, 60 seconds).get(appName).map(t => (t._2, t._1))
   }
 
-  override def getBinaryContent(appName: String, binaryType: BinaryType,
-                                uploadTime: DateTime): Array[Byte] = {
-    binaries((appName, binaryType, uploadTime))
-  }
-
-  override def deleteBinary(appName: String): Unit = {
+override def deleteBinary(appName: String): Unit = {
     binaries = binaries.filter { case ((name, _, _), _) => appName != name }
   }
 }

job-server/src/test/scala/spark/jobserver/io/JobCassandraDAOSpec.scala

@@ -137,19 +137,6 @@ class JobCassandraDAOSpec extends TestJarFinder with FunSpecLike with Matchers w
       jarFile.length() should equal (retrieved.length())
       Files.toByteArray(jarFile) should equal(Files.toByteArray(retrieved))
     }
-
-    it("should retrieve the jar binary content for remote job manager") {
-      // chack the pre-condition
-      jarFile.exists() should equal (false)
-
-      // retrieve the jar content
-      val jarBinaryContent: Array[Byte] = dao.getBinaryContent(jarInfo.appName, jarInfo.binaryType, jarInfo.uploadTime)
-
-      // test
-      jarFile.exists() should equal (true)
-      jarBinaryContent.length should equal (jarBytes.length)
-      jarBinaryContent should equal(jarBytes)
-    }
   }
 
   describe("saveJobConfig() tests") {

job-server/src/test/scala/spark/jobserver/io/JobDAOActorSpec.scala

@@ -22,7 +22,6 @@ object JobDAOActorSpec {
   val cleanupProbe = TestProbe()(system)
 
   object DummyDao extends JobDAO{
-    val jarContent = Array.empty[Byte]
 
     override def saveBinary(appName: String, binaryType: BinaryType,
                             uploadTime: DateTime, binaryBytes: Array[Byte]): Unit = {
@@ -38,14 +37,6 @@ object JobDAOActorSpec {
         "app2" -> (BinaryType.Egg, dtplus1)
       ))
 
-    override def getBinaryContent(appName: String, binaryType: BinaryType,
-                                  uploadTime: DateTime): Array[Byte] = {
-      appName match {
-        case "failOnThis" => throw new Exception("get binary content failure")
-        case _ => jarContent
-      }
-    }
-
     override def retrieveBinaryFile(appName: String,
                                     binaryType: BinaryType, uploadTime: DateTime): String = ???
 
@@ -128,11 +119,6 @@ class JobDAOActorSpec extends TestKit(JobDAOActorSpec.system) with ImplicitSende
       expectMsg(JobInfos(Seq()))
     }
 
-    it("should get binary content") {
-      daoActor ! GetBinaryContent("succeed", BinaryType.Jar, DateTime.now)
-      expectMsg(BinaryContent(DummyDao.jarContent))
-    }
-
     it("should request jobs cleanup") {
       daoActor ! CleanContextJobInfos("context", DateTime.now())
       cleanupProbe.expectMsg("context")

job-server/src/test/scala/spark/jobserver/io/JobSqlDAOSpec.scala

@@ -133,18 +133,6 @@ class JobSqlDAOSpec extends JobSqlDAOSpecBase with TestJarFinder with FunSpecLik
       jarFile.exists() should equal (true)
       jarFilePath should equal (jarFile.getAbsolutePath)
     }
-
-    it("should retrieve the jar binary content for remote job manager") {
-      // chack the pre-condition
-      jarFile.exists() should equal (false)
-
-      // retrieve the jar content
-      val jarBinaryContent: Array[Byte] = dao.getBinaryContent(jarInfo.appName, BinaryType.Jar, jarInfo.uploadTime)
-
-      // test
-      jarFile.exists() should equal (true)
-      jarBinaryContent should equal (jarBytes)
-    }
   }
 
   describe("save and get Python eggs") {