From 44d0e3ba1e0adbe221e4f32fcf7a3652c8728a9e Mon Sep 17 00:00:00 2001
From: Kevin Wallimann <kevin.wallimann@absa.africa>
Date: Wed, 6 Mar 2024 10:33:36 +0100
Subject: [PATCH 1/3] Add saveDiagnostics application property

---
 src/main/resources/application.properties                      | 1 +
 .../trigger/configuration/application/SparkConfig.scala        | 3 +++
 .../trigger/scheduler/executors/spark/SparkExecutor.scala      | 3 ++-
 .../configuration/application/DefaultTestSparkConfig.scala     | 3 +++
 4 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
index 33c5d120d..cddf30e91 100644
--- a/src/main/resources/application.properties
+++ b/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=
diff --git a/src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfig.scala b/src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfig.scala
index cafca5b54..1ab0e538b 100644
--- a/src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfig.scala
+++ b/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
diff --git a/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala b/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
index 9aebc60a6..ac88bcc33 100644
--- a/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
+++ b/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
@@ -72,9 +72,10 @@ object SparkExecutor {
   private def getUpdatedJobInstance(
     jobInstance: JobInstance,
     app: App
-  ): JobInstance = {
+  )(implicit sparkConfig: SparkConfig): JobInstance = {
     val diagnostics = app.diagnostics match {
       case "" => None
+      case _ if !sparkConfig.saveDiagnostics => None
       case _  => Some(app.diagnostics)
     }
 
diff --git a/src/test/scala/za/co/absa/hyperdrive/trigger/configuration/application/DefaultTestSparkConfig.scala b/src/test/scala/za/co/absa/hyperdrive/trigger/configuration/application/DefaultTestSparkConfig.scala
index a0f286982..fba296b5d 100644
--- a/src/test/scala/za/co/absa/hyperdrive/trigger/configuration/application/DefaultTestSparkConfig.scala
+++ b/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
     )
 

From ea38489bb8178483031239fccb1a9afd2c796d4c Mon Sep 17 00:00:00 2001
From: Kevin Wallimann <kevin.wallimann@absa.africa>
Date: Wed, 6 Mar 2024 11:11:48 +0100
Subject: [PATCH 2/3] Fix format

---
 .../trigger/scheduler/executors/spark/SparkExecutor.scala     | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala b/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
index ac88bcc33..62b058e77 100644
--- a/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
+++ b/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
@@ -74,9 +74,9 @@ object SparkExecutor {
     app: App
   )(implicit sparkConfig: SparkConfig): JobInstance = {
     val diagnostics = app.diagnostics match {
-      case "" => None
+      case ""                                => None
       case _ if !sparkConfig.saveDiagnostics => None
-      case _  => Some(app.diagnostics)
+      case _                                 => Some(app.diagnostics)
     }
 
     jobInstance.copy(

From 5af95cd10c85559c7e439f63d32b7753536d95e6 Mon Sep 17 00:00:00 2001
From: Kevin Wallimann <kevin.wallimann@absa.africa>
Date: Wed, 6 Mar 2024 14:26:01 +0100
Subject: [PATCH 3/3] PR fix

---
 .../trigger/scheduler/executors/spark/SparkExecutor.scala   | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala b/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
index 62b058e77..7126e475d 100644
--- a/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
+++ b/src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkExecutor.scala
@@ -74,9 +74,9 @@ object SparkExecutor {
     app: App
   )(implicit sparkConfig: SparkConfig): JobInstance = {
     val diagnostics = app.diagnostics match {
-      case ""                                => None
-      case _ if !sparkConfig.saveDiagnostics => None
-      case _                                 => Some(app.diagnostics)
+      case ""                               => None
+      case _ if sparkConfig.saveDiagnostics => Some(app.diagnostics)
+      case _                                => None
     }
 
     jobInstance.copy(
