Merge pull request #276 from sowmya-dixit/release-2.6.0

Fix for report jobs not exiting
project-sunbird · Jan 2, 2020 · e06b2d9 · e06b2d9
2 parents 5e0cf7e + 7c38e36
commit e06b2d9
Show file tree

Hide file tree

Showing 6 changed files with 10 additions and 3 deletions.
diff --git a/platform-api/analytics-api-core/src/test/scala/org/ekstep/analytics/api/BaseSpec.scala b/platform-api/analytics-api-core/src/test/scala/org/ekstep/analytics/api/BaseSpec.scala
@@ -20,7 +20,7 @@ class BaseSpec extends FlatSpec with Matchers with BeforeAndAfterAll with Mockit
   override def beforeAll() {
     if (embeddedCassandraMode) {
       System.setProperty("cassandra.unsafesystem", "true")
-		  EmbeddedCassandraServerHelper.startEmbeddedCassandra(20000L)
+		  EmbeddedCassandraServerHelper.startEmbeddedCassandra(30000L)
 		  val session = DBUtil.session
 		  val dataLoader = new CQLDataLoader(session);
 		  dataLoader.load(new FileCQLDataSet(AppConf.getConfig("cassandra.cql_path"), true, true));

diff --git a/...-api/analytics-api-core/src/test/scala/org/ekstep/analytics/api/util/TestCommonUtil.scala b/...-api/analytics-api-core/src/test/scala/org/ekstep/analytics/api/util/TestCommonUtil.scala
@@ -1,5 +1,6 @@
 package org.ekstep.analytics.api.util
 
+import org.ekstep.analytics.api.util.CommonUtil.monthPeriod
 import org.ekstep.analytics.api.{BaseSpec, Range, ResponseCode}
 import org.joda.time.format.{DateTimeFormat, DateTimeFormatter}
 import org.joda.time.{DateTime, DateTimeZone, Duration}
@@ -14,7 +15,7 @@ class TestCommonUtil extends BaseSpec {
         val thisYear: Int = (new DateTime).getYear
         val thisMonth: Int = (new DateTime).getMonthOfYear
         val thisWeekNo = (new DateTime).getWeekOfWeekyear
-        val monthYear = Integer.parseInt(s"$thisYear"+s"$thisMonth")
+        val monthYear = monthPeriod.print(new DateTime).toInt
 
         val now = DateTime.now(DateTimeZone.UTC)
         val dateInt = Integer.parseInt(dateFormat.print(now).replace("-", ""))
@@ -32,7 +33,8 @@ class TestCommonUtil extends BaseSpec {
         resp3.params.status should be("successful");
         resp3.result.get should be(Map("ttl" -> 24.asInstanceOf[AnyRef]));
         CommonUtil.getDayRange(7) should be(Range(Integer.parseInt(dateFormat.print(now.minusDays(7)).replace("-", "")), dateInt))
-        CommonUtil.getMonthRange(2) should be(Range(monthYear-2,  monthYear))
+        val startMonth = monthPeriod.print((new DateTime).minusMonths(2)).toInt
+        CommonUtil.getMonthRange(2) should be(Range(startMonth,  monthYear))
         CommonUtil.getRemainingHours() should be(remainingTime)
         CommonUtil.getWeeksBetween(1451650400000L, 1454650400000L) should be(5);
 

diff --git a/...es/batch-models/src/main/scala/org/ekstep/analytics/job/report/AssessmentMetricsJob.scala b/...es/batch-models/src/main/scala/org/ekstep/analytics/job/report/AssessmentMetricsJob.scala
@@ -26,6 +26,7 @@ object AssessmentMetricsJob extends optional.Application with IJob with BaseRepo
 
   def main(config: String)(implicit sc: Option[SparkContext] = None, fc: Option[FrameworkContext] = None) {
 
+
     JobLogger.init("Assessment Metrics")
     JobLogger.start("Assessment Job Started executing", Option(Map("config" -> config, "model" -> name)))
     val jobConfig = JSONUtils.deserialize[JobConfig](config)
@@ -281,6 +282,7 @@ object AssessmentMetricsJob extends optional.Application with IJob with BaseRepo
     FileUtil.renameReport(tempDir, renamedDir, batchId)
     val storageService = getReportStorageService();
     storageService.upload(container, renamedDir, objectKey, isDirectory = Option(true))
+    storageService.closeContext()
   }
 
   def saveToElastic(index: String, reportDF: DataFrame): Unit = {

diff --git a/...odules/batch-models/src/main/scala/org/ekstep/analytics/job/report/CourseMetricsJob.scala b/...odules/batch-models/src/main/scala/org/ekstep/analytics/job/report/CourseMetricsJob.scala
@@ -337,6 +337,7 @@ object CourseMetricsJob extends optional.Application with IJob with ReportGenera
     val objectKey = AppConf.getConfig("course.metrics.cloud.objectKey")
     val storageService = getReportStorageService();
     storageService.upload(container, sourcePath, objectKey, isDirectory = Option(true))
+    storageService.closeContext()
   }
 
   private def recursiveListFiles(file: File, ext: String): Array[File] = {

diff --git a/.../batch-models/src/main/scala/org/ekstep/analytics/job/report/StateAdminGeoReportJob.scala b/.../batch-models/src/main/scala/org/ekstep/analytics/job/report/StateAdminGeoReportJob.scala
@@ -106,6 +106,7 @@ object StateAdminGeoReportJob extends optional.Application with IJob with StateA
 
     val storageService = getReportStorageService();
     storageService.upload(container, sourcePath, objectKey, isDirectory = Option(true))
+    storageService.closeContext()
   }
 }
 

diff --git a/...les/batch-models/src/main/scala/org/ekstep/analytics/job/report/StateAdminReportJob.scala b/...les/batch-models/src/main/scala/org/ekstep/analytics/job/report/StateAdminReportJob.scala
@@ -254,6 +254,7 @@ object StateAdminReportJob extends optional.Application with IJob with StateAdmi
 
         val storageService = getReportStorageService();
         storageService.upload(container, sourcePath, objectKey, isDirectory = Option(true))
+        storageService.closeContext()
     }
 
     def dataFrameToJsonFile(dataFrame: DataFrame)(implicit fc: FrameworkContext): Unit = {