#401 - fixed executablesFolder for shell scripts. (#413)

Author: jozefbakus

src/main/resources/application.properties

@@ -50,7 +50,6 @@ scheduler.jobs.parallel.number=100
 scheduler.heart.beat=5000
 scheduler.lag.threshold=20000
 scheduler.sensors.changedSensorsChunkQuerySize=100
-scheduler.executors.executablesFolder=/
 
 #Kafka sensor properties.
 kafkaSource.group.id.prefix=hyper_drive
@@ -70,6 +69,9 @@ kafkaSource.properties.sasl.kerberos.service.name=
 kafkaSource.properties.sasl.mechanism=
 kafkaSource.properties.sasl.jaas.config=
 
+#Shell executor properties.
+shellExecutor.executablesFolder=/
+
 #Spark yarn sink properties. Properties used to deploy and run Spark job in Yarn.
 sparkYarnSink.hadoopResourceManagerUrlBase=
 sparkYarnSink.hadoopConfDir=
@@ -88,6 +90,7 @@ sparkYarnSink.additionalConfs.spark.yarn.keytab=
 sparkYarnSink.additionalConfs.spark.yarn.principal=
 sparkYarnSink.additionalConfs.spark.shuffle.service.enabled=true
 sparkYarnSink.additionalConfs.spark.dynamicAllocation.enabled=true
+sparkYarnSink.executablesFolder=/
 
 #Postgresql properties for connection to trigger metastore
 db.driver=net.bull.javamelody.JdbcDriver

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/shell/ShellParameters.scala

@@ -18,14 +18,14 @@ package za.co.absa.hyperdrive.trigger.scheduler.executors.shell
 import java.nio.file.Paths
 
 import za.co.absa.hyperdrive.trigger.models.JobParameters
-import za.co.absa.hyperdrive.trigger.scheduler.utilities.ExecutorsConfig
+import za.co.absa.hyperdrive.trigger.scheduler.utilities.ShellExecutorConfig
 
 case class ShellParameters(
   scriptLocation: String
 )
 
 object ShellParameters {
   def apply(jobParameters: JobParameters): ShellParameters = new ShellParameters(
-    scriptLocation = Paths.get(ExecutorsConfig.getExecutablesFolder, jobParameters.variables("scriptLocation")).toString
+    scriptLocation = Paths.get(ShellExecutorConfig.getExecutablesFolder, jobParameters.variables("scriptLocation")).toString
   )
 }

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

@@ -18,7 +18,7 @@ package za.co.absa.hyperdrive.trigger.scheduler.executors.spark
 import java.nio.file.Paths
 
 import za.co.absa.hyperdrive.trigger.models.JobParameters
-import za.co.absa.hyperdrive.trigger.scheduler.utilities.ExecutorsConfig
+import za.co.absa.hyperdrive.trigger.scheduler.utilities.SparkExecutorConfig
 
 import scala.util.Try
 
@@ -38,12 +38,12 @@ case class SparkParameters(
 object SparkParameters {
   def apply(jobParameters: JobParameters): SparkParameters = {
     SparkParameters(
-      jobJar = Paths.get(ExecutorsConfig.getExecutablesFolder, jobParameters.variables("jobJar")).toString,
+      jobJar = Paths.get(SparkExecutorConfig.getExecutablesFolder, jobParameters.variables("jobJar")).toString,
       mainClass = jobParameters.variables("mainClass"),
       deploymentMode = jobParameters.variables("deploymentMode"),
       appArguments = Try(jobParameters.maps("appArguments")).getOrElse(List.empty[String]),
-      additionalJars = Try(jobParameters.maps("additionalJars")).getOrElse(List.empty[String]).map(jar => Paths.get(ExecutorsConfig.getExecutablesFolder, jar).toString),
-      additionalFiles = Try(jobParameters.maps("additionalFiles")).getOrElse(List.empty[String]).map(file => Paths.get(ExecutorsConfig.getExecutablesFolder, file).toString),
+      additionalJars = Try(jobParameters.maps("additionalJars")).getOrElse(List.empty[String]).map(jar => Paths.get(SparkExecutorConfig.getExecutablesFolder, jar).toString),
+      additionalFiles = Try(jobParameters.maps("additionalFiles")).getOrElse(List.empty[String]).map(file => Paths.get(SparkExecutorConfig.getExecutablesFolder, file).toString),
       additionalSparkConfig = Try(jobParameters.keyValuePairs("additionalSparkConfig")).getOrElse(Map.empty[String, String])
     )
   }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/utilities/Configs.scala

@@ -88,8 +88,11 @@ object SchedulerConfig {
 object ExecutorsConfig {
   val getThreadPoolSize: Int =
     Configs.conf.getInt("scheduler.executors.thread.pool.size")
+}
+
+object ShellExecutorConfig {
   val getExecutablesFolder: String =
-    Configs.conf.getString("scheduler.executors.executablesFolder")
+    Configs.conf.getString("shellExecutor.executablesFolder")
 }
 
 object SparkExecutorConfig {
@@ -107,6 +110,8 @@ object SparkExecutorConfig {
     Try(Configs.conf.getString("sparkYarnSink.filesToDeploy").split(",").toSeq).getOrElse(Seq.empty[String])
   val getAdditionalConfs: Map[String, String] =
     Configs.getMapFromConf("sparkYarnSink.additionalConfs")
+  val getExecutablesFolder: String =
+    Configs.conf.getString("sparkYarnSink.executablesFolder")
 }
 
 object JobDefinitionConfig {

src/test/resources/application.properties

@@ -20,9 +20,12 @@ auth.inmemory.password=password
 
 appUniqueId=9c282190-4078-4380-8960-ce52f43b94fg
 
-scheduler.executors.executablesFolder=src/test/resources/
 scheduler.sensors.changedSensorsChunkQuerySize=100
 
+shellExecutor.executablesFolder=src/test/resources/
+
+sparkYarnSink.executablesFolder=src/test/resources/
+
 kafkaSource.group.id.prefix=hyper_drive
 kafkaSource.properties.security.protocol=PLAINTEXT