#738: Resurrect scheduler (#740)

* #738: Resurrect scheduler

Author: jozefbakus

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/JobScheduler.scala

@@ -84,8 +84,7 @@ class JobScheduler @Inject() (
   def stopManager(): Future[Unit] = {
     logger.info("Stopping Manager")
     isManagerRunningAtomic.set(false)
-    sensors.cleanUpSensors()
-    workflowBalancer.resetSchedulerInstanceId()
+    cleanUp()
     runningScheduler
   }
 
@@ -97,8 +96,8 @@ class JobScheduler @Inject() (
       runningAssignWorkflows = workflowBalancer
         .getAssignedWorkflows(runningDags.keys.map(_.workflowId).toSeq)
         .recover { case e: SchedulerInstanceAlreadyDeactivatedException =>
-          logger.error("Stopping scheduler because the instance has already been deactivated", e)
-          stopManager()
+          logger.warn("Restarting scheduler because the instance has been deactivated by other instance", e)
+          cleanUp()
           throw e
         }
         .map(_.map(_.id))
@@ -166,4 +165,8 @@ class JobScheduler @Inject() (
       }
     }
 
+  private def cleanUp(): Unit = {
+    sensors.cleanUpSensors()
+    workflowBalancer.resetSchedulerInstanceId()
+  }
 }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala

@@ -19,13 +19,18 @@ import play.api.libs.json.{JsValue, Json}
 import play.api.libs.ws.JsonBodyReadables._
 import za.co.absa.hyperdrive.trigger.api.rest.utils.WSClientProvider
 import za.co.absa.hyperdrive.trigger.configuration.application.SparkConfig
+import za.co.absa.hyperdrive.trigger.models.enums.JobStatuses
 import za.co.absa.hyperdrive.trigger.models.enums.JobStatuses._
 import za.co.absa.hyperdrive.trigger.models.{JobInstance, SparkInstanceParameters}
 import za.co.absa.hyperdrive.trigger.scheduler.executors.spark.{FinalStatuses => YarnFinalStatuses}
 
+import java.time.LocalDateTime
+import java.time.temporal.ChronoUnit
 import scala.concurrent.{ExecutionContext, Future}
 
 object SparkExecutor {
+  private val ExtraSubmitTimeout = 60000
+
   def execute(
     jobInstance: JobInstance,
     jobParameters: SparkInstanceParameters,
@@ -50,6 +55,13 @@ object SparkExecutor {
       }) match {
         case Seq(first) =>
           updateJob(jobInstance.copy(applicationId = Some(first.id), jobStatus = getStatus(first.finalStatus)))
+        case _
+            // It relies on the same value set for sparkYarnSink.submitTimeout in multi instance deployment
+            if jobInstance.jobStatus == JobStatuses.Submitting && jobInstance.updated
+              .map(lastUpdated => ChronoUnit.MILLIS.between(lastUpdated, LocalDateTime.now()))
+              .exists(_ < sparkConfig.yarn.submitTimeout + ExtraSubmitTimeout) =>
+          // Do nothing for submit timeout period to avoid two parallel job submissions/executions
+          Future((): Unit)
         case _ => sparkClusterService.handleMissingYarnStatus(jobInstance, updateJob)
       }
     }