Repository: AbsaOSS/pramen (language: python)

Issue:
Allow transformations to set a warning flag to indicate that the pipeline has finished with warnings
## Background
A postprocessing step can lead to errors or warnings in the dataset. 

Errors are easy - an exception can be thrown.

But warnings are not possible to add at the moment.

## Feature
Allow transformations to set a warning flag to indicate that the pipeline has finished with warnings.

## Example
--

## Proposed Solution [Optional]
Add a warning flag and give access to it from `Pramen` trait, e.g.:

```scala
Pramen.instance.setWarningFlag()
```


Discussion / hints:

Fix patch:
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
index 2d416c70e..5b8828be2 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/Pramen.scala
@@ -57,6 +57,13 @@ trait Pramen {
     * Returns the list of tasks completed at the current moment of time.
     */
   def getCompletedTasks: Seq[TaskResult]
+
+  /**
+    * Sets the warning flag, which means that the pipeline will be reported as succeeded with warnings.
+    *
+    * This can be set from any operation, but cannot be unset.
+    */
+  def setWarningFlag(): Unit
 }
 
 object Pramen {
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala
index 0f2941b1c..1f98496b2 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/status/PipelineStateSnapshot.scala
@@ -21,6 +21,7 @@ import za.co.absa.pramen.api.PipelineInfo
 case class PipelineStateSnapshot(
                                   pipelineInfo: PipelineInfo,
                                   isFinished: Boolean,
+                                  warningFlag: Boolean,
                                   exitedNormally: Boolean,
                                   exitCode: Int,
                                   customShutdownHookCanRun: Boolean,
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
index d0c04ff0e..f70c08806 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/PramenImpl.scala
@@ -71,6 +71,14 @@ class PramenImpl extends Pramen {
     state.taskResults
   }
 
+  override def setWarningFlag(): Unit = {
+    val pipelineState = _pipelineState.getOrElse(
+      throw new IllegalStateException("Pipeline state is not available at the context.")
+    )
+
+    pipelineState.setWarningFlag()
+  }
+
   private[core] def setWorkflowConfig(config: Config): Unit = synchronized {
     _workflowConfig = Option(config)
   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala
index 00609b024..3de9f1758 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala
@@ -23,6 +23,7 @@ import java.time.Instant
 
 case class PipelineNotification(
                                  exception: Option[Throwable],
+                                 warningFlag: Boolean,
                                  pipelineName: String,
                                  environmentName: String,
                                  sparkAppId: Option[String],
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala
index f1417c00e..149a0294a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala
@@ -24,6 +24,8 @@ import java.time.Instant
 trait PipelineNotificationBuilder {
   def addFailureException(ex: Throwable): Unit
 
+  def addWarningFlag(warningFlag: Boolean): Unit
+
   def addAppName(appName: String): Unit
 
   def addSparkAppId(sparkAppId: String): Unit
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
index b34fc970b..f447b1f02 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
@@ -53,6 +53,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
   private var goodRps = Int.MaxValue
 
   var appException: Option[Throwable] = None
+  var warningFlag: Boolean = false
   var appName: String = "Unspecified Job"
   var sparkAppId: Option[String] = None
   var envName: String = "Unspecified Environment"
@@ -70,6 +71,10 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     appException = Option(ex)
   }
 
+  override def addWarningFlag(flag: Boolean): Unit = {
+    warningFlag = flag
+  }
+
   override def addAppName(appName: String): Unit = {
     this.appName = appName
   }
@@ -165,7 +170,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
   def pipelineStatus: PipelineStatus = {
     val isCertainFailure = appException.nonEmpty
     val (someTasksSucceeded, someTasksFailed) = getSuccessFlags
-    val hasAtLeastOneWarning = hasWarnings
+    val hasAtLeastOneWarning = warningFlag || hasWarnings
 
     if (isCertainFailure) {
       PipelineStatus.Failure
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationDirector.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationDirector.scala
index 0d526c795..62a3ad05a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationDirector.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationDirector.scala
@@ -42,6 +42,7 @@ object PipelineNotificationDirector {
     notificationBuilder.addUndercover(undercover)
 
     notification.exception.foreach(notificationBuilder.addFailureException)
+    notificationBuilder.addWarningFlag(notification.warningFlag)
 
     notificationBuilder.addRpsMetrics(minRps, goodRps)
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala
index 8057284ea..075c6dd08 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineState.scala
@@ -25,6 +25,8 @@ trait PipelineState extends AutoCloseable {
 
   def setSuccess(): Unit
 
+  def setWarningFlag(): Unit
+
   def setFailure(stage: String, exception: Throwable): Unit
 
   def setSparkAppId(sparkAppId: String): Unit
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
index d85159d93..b5dcf6b7a 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
@@ -61,6 +61,7 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
   @volatile private var isFinished = false
   @volatile private var customShutdownHookCanRun = false
   @volatile private var sparkAppId: Option[String] = None
+  @volatile private var warningFlag: Boolean = false
 
   init()
 
@@ -110,6 +111,7 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
         pipelineNotificationFailures.toSeq
       ),
       isFinished,
+      warningFlag,
       exitedNormally,
       exitCode,
       customShutdownHookCanRun,
@@ -130,6 +132,10 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
     }
   }
 
+  override def setWarningFlag(): Unit = synchronized {
+    warningFlag = true
+  }
+
   override def setFailure(stage: String, exception: Throwable): Unit = synchronized {
     if (!alreadyFinished()) {
       setFailureException(exception)
@@ -249,6 +255,7 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
       val customSignature = notificationBuilderImpl.signature
 
       val notification = PipelineNotification(failureException,
+        warningFlag,
         pipelineName,
         environmentName,
         sparkAppId,


Test patch:
diff --git a/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala b/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
index febdaf329..8f0c828ea 100644
--- a/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
+++ b/pramen/api/src/test/scala/za/co/absa/pramen/api/mocks/DummyPramen.scala
@@ -35,4 +35,6 @@ class DummyPramen extends Pramen {
   override def metadataManager: MetadataManager = null
 
   override def getCompletedTasks: Seq[TaskResult] = null
+
+  override def setWarningFlag(): Unit = null
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala
index 26f9abab9..41ed3cd36 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineNotificationFactory.scala
@@ -25,6 +25,7 @@ import java.time.Instant
 
 object PipelineNotificationFactory {
   def getDummyNotification(exception: Option[Throwable] = None,
+                           warningFlag: Boolean = false,
                            pipelineName: String = "DummyPipeline",
                            environmentName: String = "DummyEnvironment",
                            sparkAppId: Option[String] = None,
@@ -37,6 +38,7 @@ object PipelineNotificationFactory {
                           ): PipelineNotification = {
     pipeline.PipelineNotification(
       exception,
+      warningFlag,
       pipelineName,
       environmentName,
       sparkAppId,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala
index bdc1ad2c9..de1999762 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineStateSnapshotFactory.scala
@@ -22,6 +22,7 @@ import za.co.absa.pramen.api.status.{CustomNotification, PipelineNotificationFai
 object PipelineStateSnapshotFactory {
   def getDummyPipelineStateSnapshot(pipelineInfo: PipelineInfo = PipelineInfoFactory.getDummyPipelineInfo(),
                                     isFinished: Boolean = false,
+                                    warningFlag: Boolean = false,
                                     exitedNormally: Boolean = false,
                                     exitCode: Int = 0,
                                     customShutdownHookCanRun: Boolean = false,
@@ -30,6 +31,7 @@ object PipelineStateSnapshotFactory {
                                     customNotification: CustomNotification = CustomNotification(Seq.empty, Seq.empty)): PipelineStateSnapshot = {
     PipelineStateSnapshot(pipelineInfo,
       isFinished,
+      warningFlag,
       exitedNormally,
       exitCode,
       customShutdownHookCanRun,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala
index 461f7bb31..296298e91 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/notify/PipelineNotificationBuilderSpy.scala
@@ -24,6 +24,7 @@ import java.time.Instant
 
 class PipelineNotificationBuilderSpy extends PipelineNotificationBuilder {
   var failureException: Option[Throwable] = None
+  var warningFlag: Boolean = false
   var appName = ""
   var sparkId = ""
   var environmentName = ""
@@ -41,6 +42,8 @@ class PipelineNotificationBuilderSpy extends PipelineNotificationBuilder {
 
   override def addFailureException(ex: Throwable): Unit = failureException = Option(ex)
 
+  override def addWarningFlag(flag: Boolean): Unit = warningFlag = flag
+
   override def addAppName(name: String): Unit = appName = name
 
   override def addSparkAppId(sparkAppId: String): Unit = sparkId = sparkAppId
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala
index 5b199d3f4..96c0d0b4e 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/state/PipelineStateSpy.scala
@@ -25,6 +25,7 @@ import scala.collection.mutable.ListBuffer
 class PipelineStateSpy extends PipelineState {
   var setShutdownHookCanRunCount = 0
   var setSuccessCount = 0
+  var setWarningCount = 0
   var getExitCodeCalled = 0
   var jobFailureCalled = 0
   val failures = new ListBuffer[(String, Throwable)]
@@ -47,6 +48,10 @@ class PipelineStateSpy extends PipelineState {
     setSuccessCount += 1
   }
 
+  override def setWarningFlag(): Unit = synchronized {
+    setWarningCount += 1
+  }
+
   override def setFailure(stage: String, ex: Throwable): Unit = synchronized {
     failures.append((stage, ex))
   }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala
index 0f5c9e396..b6d1f3fb7 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/notify/pipeline/PipelineNotificationDirectorSuite.scala
@@ -51,7 +51,7 @@ class PipelineNotificationDirectorSuite extends AnyWordSpec {
       val customEntries = List(NotificationEntry.Paragraph(TextElement("1") :: Nil), NotificationEntry.Paragraph(TextElement("1") :: Nil))
       val pipelineNotificationFailures = List(PipelineNotificationFailure("Dummy Reason", new IllegalStateException("Dummy")))
 
-      val notification = PipelineNotificationFactory.getDummyNotification(exception = Some(ex2), tasksCompleted =
+      val notification = PipelineNotificationFactory.getDummyNotification(exception = Some(ex2), warningFlag = true, tasksCompleted =
         List(taskCompleted1, taskCompleted2, taskCompleted3), customEntries = customEntries, pipelineNotificationFailures = pipelineNotificationFailures)
 
       val builderSpy = new PipelineNotificationBuilderSpy
@@ -63,6 +63,7 @@ class PipelineNotificationDirectorSuite extends AnyWordSpec {
       assert(builderSpy.appStarted == Instant.ofEpochSecond(1234567L))
       assert(builderSpy.appFinished == Instant.ofEpochSecond(1234568L))
       assert(builderSpy.failureException.contains(ex2))
+      assert(builderSpy.warningFlag)
       assert(builderSpy.isDryRun.contains(true))
       assert(builderSpy.isUndercover.contains(false))
       assert(builderSpy.minRps == 1000)