refactor(jobserver): removed deprecated getJobConfigs (spark-jobserver#943)

Remove unused getJobConfigs from all over. Due to
performance reasons it was deprecated and doing
new changes to some parts of *DAO require changes
to this part, which is not cool.

Author: bsikander

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

@@ -298,16 +298,6 @@ class JobCassandraDAO(config: Config) extends JobDAO with FileCacher {
     }
   }
 
-  override def getJobConfigs: Future[Map[String, Config]] = {
-    val query = QB.select(Metadata.JobId, Metadata.JobConfig).from(Metadata.JobsTable)
-    session.executeAsync(query).map { rs =>
-      JListWrapper(rs.all()).map { row =>
-        val config = Option(row.getString(Metadata.JobConfig)).getOrElse("")
-        (row.getUUID(Metadata.JobId).toString, ConfigFactory.parseString(config))
-      }.toMap
-    }
-  }
-
   override def getJobConfig(jobId: String): Future[Option[Config]] = {
     val query = QB.select(Metadata.JobConfig).from(Metadata.JobsTable)
       .where(QB.eq(Metadata.JobId, UUID.fromString(jobId)))

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

@@ -172,14 +172,6 @@ trait JobDAO {
    */
   def saveJobConfig(jobId: String, jobConfig: Config)
 
-  /**
-   * Return all job ids to their job configuration.
-   * @todo remove. used only in test
-   * @return
-   */
-  @deprecated("Leads to performance problems and OutOfMemory error ultimately", "0.7.1")
-  def getJobConfigs: Future[Map[String, Config]]
-
   /**
     * Returns a config for a given jobId
     * @return

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

@@ -36,8 +36,6 @@ object JobDAOActor {
   case class GetJobInfos(limit: Int) extends JobDAORequest
 
   case class SaveJobConfig(jobId: String, jobConfig: Config) extends JobDAORequest
-  @deprecated("Leads to performance problems and OutOfMemory error ultimately", "0.7.1")
-  case object GetJobConfigs extends JobDAORequest
   case class GetJobConfig(jobId: String) extends JobDAORequest
   case class CleanContextJobInfos(contextName: String, endTime: DateTime)
 
@@ -49,7 +47,6 @@ object JobDAOActor {
   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 JobConfigs(jobConfigs: Map[String, Config]) extends JobDAOResponse
   case class JobConfig(jobConfig: Option[Config]) extends JobDAOResponse
   case class LastUploadTimeAndType(uploadTimeAndType: Option[(DateTime, BinaryType)]) extends JobDAOResponse
 
@@ -89,9 +86,6 @@ class JobDAOActor(dao: JobDAO) extends InstrumentedActor {
     case SaveJobConfig(jobId, jobConfig) =>
       dao.saveJobConfig(jobId, jobConfig)
 
-    case GetJobConfigs =>
-      dao.getJobConfigs.map(JobConfigs).pipeTo(sender)
-
     case GetJobConfig(jobId) =>
       dao.getJobConfig(jobId).map(JobConfig).pipeTo(sender)
 

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

@@ -220,8 +220,6 @@ class JobFileDAO(config: Config) extends JobDAO {
     configs(jobId) = jobConfig
   }
 
-  override def getJobConfigs: Future[Map[String, Config]] = Future { configs.toMap }
-
   override def getJobConfig(jobId: String): Future[Option[Config]] = Future {
     configs.get(jobId)
   }

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

@@ -231,14 +231,6 @@ class JobSqlDAO(config: Config) extends JobDAO with FileCacher {
   // Convert from java.sql.Timestamp to joda DateTime
   private def convertDateSqlToJoda(timestamp: Timestamp): DateTime = new DateTime(timestamp.getTime)
 
-  override def getJobConfigs: Future[Map[String, Config]] = {
-    for (r <- db.run(configs.result)) yield {
-      r.map {
-        case (jobId, jobConfig) => jobId -> ConfigFactory.parseString(jobConfig)
-      }.toMap
-    }
-  }
-
   override def getJobConfig(jobId: String): Future[Option[Config]] = {
     val query = configs
       .filter(_.jobId === jobId).map(_.jobConfig).result

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

@@ -77,10 +77,6 @@ class InMemoryDAO extends JobDAO {
 
   override def saveJobConfig(jobId: String, jobConfig: Config) { jobConfigs(jobId) = jobConfig }
 
-  override def getJobConfigs: Future[Map[String, Config]] = Future {
-    jobConfigs.toMap
-  }
-
   override  def getJobConfig(jobId: String): Future[Option[Config]] = Future {
     jobConfigs.get(jobId)
   }

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

@@ -152,12 +152,7 @@ class JobCassandraDAOSpec extends TestJarFinder with FunSpecLike with Matchers w
     }
   }
 
-  describe("saveJobConfig() and getJobConfigs() tests") {
-    it("should provide an empty map on getJobConfigs() for an empty CONFIGS table") {
-      val configs = Await.result(dao.getJobConfigs, timeout)
-      (Map.empty[String, Config]) should equal (configs)
-    }
-
+  describe("saveJobConfig() tests") {
     it("should provide None on getJobConfig(jobId) where there is no config for a given jobId") {
       val config = Await.result(dao.getJobConfig("44c32fe1-38a4-11e1-a06a-485d60c81a3e"), timeout)
       config shouldBe None
@@ -167,26 +162,19 @@ class JobCassandraDAOSpec extends TestJarFinder with FunSpecLike with Matchers w
       // save job config
       dao.saveJobConfig(jobId, jobConfig)
 
-      // get all configs
-      val configs = Await.result(dao.getJobConfigs, timeout)
       val config = Await.result(dao.getJobConfig(jobId), timeout).get
 
       // test
-      configs.keySet should equal (Set(jobId))
-      configs(jobId) should equal (expectedConfig)
       config should equal (expectedConfig)
     }
 
     it("should be able to get previously saved config") {
       // config saved in prior test
 
       // get job configs
-      val configs = Await.result(dao.getJobConfigs, timeout)
       val config = Await.result(dao.getJobConfig(jobId), timeout).get
 
       // test
-      configs.keySet should equal (Set(jobId))
-      configs(jobId) should equal (expectedConfig)
       config should equal (expectedConfig)
     }
 
@@ -204,14 +192,10 @@ class JobCassandraDAOSpec extends TestJarFinder with FunSpecLike with Matchers w
       dao = new JobCassandraDAO(config)
 
       // Get all configs
-      val configs = Await.result(dao.getJobConfigs, timeout)
       val jobIdConfig = Await.result(dao.getJobConfig(jobId), timeout).get
       val jobId2Config = Await.result(dao.getJobConfig(jobId2), timeout).get
 
       // test
-      configs.keySet should equal (Set(jobId, jobId2))
-      configs.values.toSeq should contain (expectedConfig)
-      configs.values.toSeq should contain (expectedConfig2)
       jobIdConfig should equal (expectedConfig)
       jobId2Config should equal (expectedConfig2)
     }

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

@@ -59,8 +59,6 @@ object JobDAOActorSpec {
 
     override def saveJobInfo(jobInfo: JobInfo): Unit = ???
 
-    override def getJobConfigs: Future[Map[String, Config]] = ???
-
     override def getJobConfig(jobId: String): Future[Option[Config]] = ???
 
     override def deleteBinary(appName: String): Unit = {

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

@@ -178,11 +178,7 @@ class JobSqlDAOSpec extends JobSqlDAOSpecBase with TestJarFinder with FunSpecLik
     }
   }
 
-  describe("saveJobConfig() and getJobConfigs() tests") {
-    it("should provide an empty map on getJobConfigs() for an empty CONFIGS table") {
-      Map.empty[String, Config] should equal (Await.result(dao.getJobConfigs, timeout))
-    }
-
+  describe("saveJobConfig() tests") {
     it("should provide None on getJobConfig(jobId) where there is no config for a given jobId") {
       val config = Await.result(dao.getJobConfig("44c32fe1-38a4-11e1-a06a-485d60c81a3e"), timeout)
       config shouldBe None
@@ -193,25 +189,18 @@ class JobSqlDAOSpec extends JobSqlDAOSpecBase with TestJarFinder with FunSpecLik
       dao.saveJobConfig(jobId, jobConfig)
 
       // get all configs
-      val configs = Await.result(dao.getJobConfigs, timeout)
       val config = Await.result(dao.getJobConfig(jobId), timeout).get
 
       // test
-      configs.keySet should equal (Set(jobId))
-      configs(jobId) should equal (expectedConfig)
       config should equal (expectedConfig)
     }
 
     it("should be able to get previously saved config") {
       // config saved in prior test
 
-      // get job configs
-      val configs = Await.result(dao.getJobConfigs, timeout)
       val config = Await.result(dao.getJobConfig(jobId), timeout).get
 
       // test
-      configs.keySet should equal (Set(jobId))
-      configs(jobId) should equal (expectedConfig)
       config should equal (expectedConfig)
     }
 
@@ -229,13 +218,10 @@ class JobSqlDAOSpec extends JobSqlDAOSpecBase with TestJarFinder with FunSpecLik
       dao = new JobSqlDAO(config)
 
       // Get all configs
-      val configs = Await.result(dao.getJobConfigs, timeout)
       val jobIdConfig = Await.result(dao.getJobConfig(jobId), timeout).get
       val jobId2Config = Await.result(dao.getJobConfig(jobId2), timeout).get
 
       // test
-      configs.keySet should equal (Set(jobId, jobId2))
-      configs.values.toSeq should equal (Seq(expectedConfig, expectedConfig2))
       jobIdConfig should equal (expectedConfig)
       jobId2Config should equal (expectedConfig2)
     }