Add saveDiagnostics application property (#841)

* Add saveDiagnostics application property

* Fix format

* PR fix

Author: kevinwallimann

src/main/resources/application.properties

@@ -100,6 +100,7 @@ kafkaSource.properties.sasl.jaas.config=
 spark.submitApi=yarn
 sparkYarnSink.hadoopResourceManagerUrlBase=http://localhost:8088
 sparkYarnSink.userUsedToKillJob=
+sparkYarnSink.saveDiagnostics=false
 sparkYarnSink.master=yarn
 sparkYarnSink.submitTimeout=160000
 sparkYarnSink.filesToDeploy=

src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfig.scala

@@ -42,6 +42,9 @@ class SparkConfig(
   @DefaultValue(Array("Unknown"))
   @Name("sparkYarnSink.userUsedToKillJob")
   val userUsedToKillJob: String,
+  @DefaultValue(Array("false"))
+  @Name("sparkYarnSink.saveDiagnostics")
+  val saveDiagnostics: Boolean,
   @DefaultValue(Array("10"))
   @Name("spark.submit.thread.pool.size")
   val sparkSubmitThreadPoolSize: Int

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

@@ -72,10 +72,11 @@ object SparkExecutor {
   private def getUpdatedJobInstance(
     jobInstance: JobInstance,
     app: App
-  ): JobInstance = {
+  )(implicit sparkConfig: SparkConfig): JobInstance = {
     val diagnostics = app.diagnostics match {
-      case "" => None
-      case _  => Some(app.diagnostics)
+      case ""                               => None
+      case _ if sparkConfig.saveDiagnostics => Some(app.diagnostics)
+      case _                                => None
     }
 
     jobInstance.copy(

src/test/scala/za/co/absa/hyperdrive/trigger/configuration/application/DefaultTestSparkConfig.scala

@@ -25,6 +25,7 @@ case class DefaultTestSparkConfig(
   filesToDeploy: Seq[String] = Seq(),
   additionalConfs: Map[String, String] = Map(),
   userUsedToKillJob: String = "Unknown",
+  saveDiagnostics: Boolean = false,
   sparkSubmitThreadPoolSize: Int = 10,
   clusterId: String = "j-2AXXXXXXGAPLF"
 ) {
@@ -35,6 +36,7 @@ case class DefaultTestSparkConfig(
       null,
       hadoopResourceManagerUrlBase,
       userUsedToKillJob,
+      saveDiagnostics,
       sparkSubmitThreadPoolSize
     )
 
@@ -45,6 +47,7 @@ case class DefaultTestSparkConfig(
       new SparkEmrSinkConfig(clusterId, filesToDeploy.mkString(","), toProperties(additionalConfs)),
       hadoopResourceManagerUrlBase,
       userUsedToKillJob,
+      saveDiagnostics,
       sparkSubmitThreadPoolSize
     )