feat(jobs): add configurable autoretry

aphp · Jan 31, 2025 · 08e3280 · 08e3280
1 parent 86fea9f
commit 08e3280
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 4 deletions.
diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf
@@ -3,6 +3,8 @@ app {
     jobs {
         threads = 20
         threads = ${?JOBS_THREADS}
+        autoRetry = 0
+        autoRetry = ${?JOBS_AUTO_RETRY}
     }
     back {
       url: ${?DJANGO_CALLBACK_URL}

diff --git a/src/main/scala/fr/aphp/id/eds/requester/AppConfig.scala b/src/main/scala/fr/aphp/id/eds/requester/AppConfig.scala
@@ -33,7 +33,8 @@ case class PGConfig(
 )
 
 case class JobConfig(
-    threads: Int
+    threads: Int,
+    autoRetry: Int = 0
 )
 
 case class ServerConfig(
@@ -119,7 +120,7 @@ class AppConfig(conf: Config) {
       ))
   } else { None }
   val business: BusinessConfig = BusinessConfig(
-    JobConfig(conf.getInt("app.jobs.threads")),
+    JobConfig(conf.getInt("app.jobs.threads"), conf.getInt("app.jobs.autoRetry")),
     conf.getInt("app.cohortCreationLimit"),
     conf.getBoolean("app.enableCache"),
     QueryConfig(

diff --git a/src/main/scala/fr/aphp/id/eds/requester/jobs/JobManager.scala b/src/main/scala/fr/aphp/id/eds/requester/jobs/JobManager.scala
@@ -34,9 +34,10 @@ class JobManager() {
     sparkSession.sparkContext.addFile(AppConfig.get.solr.get.authFile)
   }
 
-  def execJob(jobExecutor: JobBase, jobData: SparkJobParameter): JobStatus = {
+  def execJob(jobExecutor: JobBase, jobData: SparkJobParameter, retry: Int = 0): JobStatus = {
     val jobId = UUID.randomUUID().toString
     logger.info(s"Starting new job ${jobId}")
+    val autoRetry = AppConfig.get.business.jobs.autoRetry
     val jobExec = Future {
       logger.info(s"Job ${jobId} started")
       sparkSession.sparkContext.setJobGroup(jobId, s"new job ${jobId}", interruptOnCancel = true)
@@ -60,7 +61,12 @@ class JobManager() {
         finalizeJob(jobId, Right(result), jobExecutor, jobData.mode, jobData)
       case Failure(wrapped: Throwable) =>
         logger.error(s"Job ${jobId} failed", wrapped)
-        finalizeJob(jobId, Left(wrapped), jobExecutor, jobData.mode, jobData)
+        if (retry < autoRetry) {
+          logger.info(s"Retrying job ${jobId}")
+          execJob(jobExecutor, jobData, retry + 1)
+        } else {
+          finalizeJob(jobId, Left(wrapped), jobExecutor, jobData.mode, jobData)
+        }
     }
     JobStatus(job.status,
               job.jobId,

diff --git a/src/test/resources/application.test.conf b/src/test/resources/application.test.conf
@@ -2,6 +2,7 @@ app {
     enableCache = false
     jobs {
         threads = 100
+        autoRetry = 0
     }
     back {
         url = "http://django"