Repository: AbsaOSS/pramen (language: python)

Issue:
Add an option that does not threat late or new data as errors separately
## Background
In case data is weekly, but the schedule is daily, pipeline is going always to fail with no data for days when the data is not available, even though it shouldn't be.

We can implement a workaround for the behavior adding a special option.

This way we can combine when the pipeline fails on new or late data separarely.

## Feature
Add an option that does not threat late data as errors.

## Example [Optional]
```hocon
sources = [ {
  name = "MySource"
  factory.class = "..."

  fail.if.no.late.data = false
  fail.if.no.new.data = true
}]
```

We already have:
```hocon
sources = [ {
  name = "MySource"
  factory.class = "..."

  fail.if.no.data = false
}]
```
which affects all data. But the proposal is to create an option applicable for the late data only.

## Proposed Solution [Optional]
Solution Ideas 
1. 
2. 
3.



Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 90c1a94de..fc8508d0f 100644
--- a/README.md
+++ b/README.md
@@ -465,7 +465,15 @@ pramen.sources = [
     format = "parquet"
     
     minimum.records = 0
+
+    # If true, fails the pipeline is there is no data any time when it is expected
     fail.if.no.data = false
+
+    # If true, fails the pipeline is there is no data for jobs trying to catch late data
+    fail.if.no.late.data = false
+
+    # If true, fails the pipeline is there is no data for jobs checking new data as expected
+    fail.if.no.new.data = false
   }
 ]
 ```
@@ -534,10 +542,16 @@ is determined by the pipeline configuration.
     # SQL queries that do not start with "SELECT".
     use.jdbc.native = false
     
-    # Consider the pipeline as failed if at least one table has no data at the scheduled time.
+    # Consider the pipeline as failed if at least one table has no data at the scheduled time (new or late).
     # Useful for auto-retrying ingestion pipelines.
     fail.if.no.data = false
 
+    # If true, fails the pipeline is there is no data for jobs trying to catch late data
+    fail.if.no.late.data = false
+
+    # If true, fails the pipeline is there is no data for jobs checking new data as expected
+    fail.if.no.new.data = false
+
     # One of: auto (default), always, never 
     # - When 'auto', an identifier will be quoted if it contains invalid characters. This includes any characters 
     #   outside the scope of A-Z, a-z, 0-9, and underscore (_).
@@ -1952,7 +1966,7 @@ Here is an example configuration for a JDBC source:
       # You can override any of source settings here 
       source {
         minimum.records = 1000 
-        fail.if.no.data = true
+        fail.if.no.new.data = true
         has.information.date.column = true
         use.jdbc.native = true
         information.date.column = "info_date"
diff --git a/pramen/core/src/main/resources/reference.conf b/pramen/core/src/main/resources/reference.conf
index 578abc155..547a7f73d 100644
--- a/pramen/core/src/main/resources/reference.conf
+++ b/pramen/core/src/main/resources/reference.conf
@@ -92,10 +92,6 @@ pramen {
 
   use.lock = true
 
-  # Consider the pipeline as failed if at least one table has no data at the scheduled time.
-  # Useful for auto-retrying ingestion pipelines.
-  fail.if.no.data = false
-
   # Send an email even if there are no changes and no late or not ready data
   email.if.no.changes = true
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala
index f73509435..7b0cc0b8f 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/IngestionJob.scala
@@ -19,7 +19,7 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.Config
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.{DataFrame, SparkSession}
-import za.co.absa.pramen.api.status.DependencyWarning
+import za.co.absa.pramen.api.status.{DependencyWarning, TaskRunReason}
 import za.co.absa.pramen.api.{Query, Reason, Source, SourceResult}
 import za.co.absa.pramen.core.app.config.GeneralConfig.TEMPORARY_DIRECTORY_KEY
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
@@ -65,11 +65,19 @@ class IngestionJob(operationDef: OperationDef,
       0
   }
 
-  override def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
+  override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
     source.connect()
 
     val minimumRecordsOpt = ConfigUtils.getOptionInt(source.config, MINIMUM_RECORDS_KEY)
-    val failIfNoData = ConfigUtils.getOptionBoolean(source.config, FAIL_NO_DATA_KEY).getOrElse(false)
+    val failIfNoAnyData = ConfigUtils.getOptionBoolean(source.config, FAIL_NO_DATA_KEY).getOrElse(false)
+    val failIfNoNewData = ConfigUtils.getOptionBoolean(source.config, FAIL_NO_NEW_DATA_KEY).getOrElse(false)
+    val failIfNoLateData = ConfigUtils.getOptionBoolean(source.config, FAIL_NO_LATE_DATA_KEY).getOrElse(false)
+
+    val failIfNoData = if (runReason == TaskRunReason.Late) {
+      failIfNoAnyData || failIfNoLateData
+    } else {
+      failIfNoAnyData || failIfNoNewData
+    }
 
     val dataChunk = bookkeeper.getLatestDataChunk(sourceTable.metaTableName, infoDate, infoDate)
 
@@ -85,7 +93,8 @@ class IngestionJob(operationDef: OperationDef,
         log.warn(s"Validation of '${outputTable.name}' for $from..$to returned warnings: ${warnings.mkString("\n")}.")
         warnings
       case Reason.NotReady(msg) =>
-        log.warn(s"Validation of '${outputTable.name}' for $from..$to returned 'not ready'. Reason: $msg.")
+        log.info(s"failIfNoAnyData=$failIfNoAnyData, failIfNoLateData=$failIfNoLateData, failIfNoNewData=$failIfNoNewData, failIfNoData=$failIfNoData")
+        log.warn(s"Validation of '${outputTable.name}' for $from..$to returned 'not ready'. Reason: $msg. ")
         return JobPreRunResult(JobPreRunStatus.NoData(failIfNoData), None, dependencyWarnings, Seq(msg))
       case Reason.Skip(msg) =>
         log.warn(s"Validation of '${outputTable.name}' for $from..$to requested to skip the task. Reason: $msg.")
@@ -229,7 +238,7 @@ class IngestionJob(operationDef: OperationDef,
         log.info(s"Table '${outputTable.name}' for $infoDate has not enough records. Minimum $minimumRecords, got $recordCount. Skipping...")
         JobPreRunResult(JobPreRunStatus.InsufficientData(recordCount, minimumRecords, oldRecordCount), minimumRecordsOpt.map(_ => recordCount), dependencyWarnings, Seq.empty[String])
       case None                 =>
-        log.info(s"Table '${outputTable.name}' for $infoDate has no data. Skipping...")
+        log.info(s"Table '${outputTable.name}' for $infoDate has no data (failIfNoData=$failIfNoData). Skipping...")
         JobPreRunResult(JobPreRunStatus.NoData(failIfNoData), minimumRecordsOpt.map(_ => recordCount), dependencyWarnings, Seq.empty[String])
     }
   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Job.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Job.scala
index 896f1f19a..04ede107e 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Job.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Job.scala
@@ -19,6 +19,7 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.Config
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.types.StructType
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.api.{Query, Reason}
 import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.runner.splitter.ScheduleStrategy
@@ -44,6 +45,7 @@ trait Job {
     * Checks pre-conditions for the job, such as data availability.
     */
   def preRunCheck(infoDate: LocalDate,
+                  runReason: TaskRunReason,
                   conf: Config): JobPreRunResult
 
   /**
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
index a70fbbb7c..ad14fa81f 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
@@ -19,7 +19,7 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.Config
 import org.apache.spark.sql.types.StructType
 import org.slf4j.{Logger, LoggerFactory}
-import za.co.absa.pramen.api.status.{DependencyFailure, DependencyWarning, MetastoreDependency}
+import za.co.absa.pramen.api.status.{DependencyFailure, DependencyWarning, MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.expr.DateExprEvaluator
 import za.co.absa.pramen.core.metastore.Metastore
@@ -50,9 +50,10 @@ abstract class JobBase(operationDef: OperationDef,
 
   override def trackDays: Int = outputTable.trackDays
 
-  def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult
+  def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult
 
   final override def preRunCheck(infoDate: LocalDate,
+                                 runReason: TaskRunReason,
                                  conf: Config): JobPreRunResult = {
     val validationFailures = operationDef.dependencies.flatMap(dependency => {
       checkDependency(dependency, infoDate)
@@ -77,7 +78,7 @@ abstract class JobBase(operationDef: OperationDef,
         log.info(s"Job for table ${outputTableDef.name} at $infoDate has no validation failures.")
       }
 
-      preRunCheckJob(infoDate, conf, dependencyWarnings)
+      preRunCheckJob(infoDate, runReason, conf, dependencyWarnings)
     }
   }
 
@@ -224,5 +225,7 @@ abstract class JobBase(operationDef: OperationDef,
 object JobBase {
   val MINIMUM_RECORDS_KEY = "minimum.records"
   val FAIL_NO_DATA_KEY = "fail.if.no.data"
+  val FAIL_NO_LATE_DATA_KEY = "fail.if.no.late.data"
+  val FAIL_NO_NEW_DATA_KEY = "fail.if.no.new.data"
   val MINIMUM_RECORDS_DEFAULT = 1
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala
index e12036534..7dfbfefdd 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJob.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{AnalysisException, DataFrame, SparkSession}
-import za.co.absa.pramen.api.status.DependencyWarning
+import za.co.absa.pramen.api.status.{DependencyWarning, TaskRunReason}
 import za.co.absa.pramen.api.{DataFormat, Reason}
 import za.co.absa.pramen.core.app.config.GeneralConfig.TEMPORARY_DIRECTORY_KEY
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
@@ -71,7 +71,7 @@ class PythonTransformationJob(operationDef: OperationDef,
 
   override val scheduleStrategy: ScheduleStrategy = new ScheduleStrategySourcing
 
-  override def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
+  override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
     validateTransformationAlreadyRanCases(infoDate, dependencyWarnings) match {
       case Some(result) => result
       case None => JobPreRunResult(JobPreRunStatus.Ready, None, dependencyWarnings, Seq.empty[String])
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
index f904812bf..a055f1aaf 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkJob.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{DataFrame, SparkSession}
-import za.co.absa.pramen.api.status.DependencyWarning
+import za.co.absa.pramen.api.status.{DependencyWarning, TaskRunReason}
 import za.co.absa.pramen.api.{Reason, Sink}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.model.MetaTable
@@ -47,7 +47,7 @@ class SinkJob(operationDef: OperationDef,
 
   override val scheduleStrategy: ScheduleStrategy = new ScheduleStrategySourcing
 
-  override def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
+  override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
     val alreadyRanStatus = preRunTransformationCheck(infoDate, dependencyWarnings)
 
     alreadyRanStatus.status match {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala
index f2c7182e4..869414b2b 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferJob.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{DataFrame, SparkSession}
-import za.co.absa.pramen.api.status.DependencyWarning
+import za.co.absa.pramen.api.status.{DependencyWarning, TaskRunReason}
 import za.co.absa.pramen.api.{Reason, Sink, Source}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.Metastore
@@ -47,8 +47,8 @@ class TransferJob(operationDef: OperationDef,
 
   override val scheduleStrategy: ScheduleStrategy = ingestionJob.scheduleStrategy
 
-  override def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
-    ingestionJob.preRunCheckJob(infoDate, jobConfig, dependencyWarnings)
+  override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
+    ingestionJob.preRunCheckJob(infoDate, runReason, jobConfig, dependencyWarnings)
   }
 
   override def validate(infoDate: LocalDate, jobConfig: Config): Reason = {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
index 78f7622f1..7c0750d2b 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransformationJob.scala
@@ -18,7 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.apache.spark.sql.{DataFrame, SparkSession}
-import za.co.absa.pramen.api.status.DependencyWarning
+import za.co.absa.pramen.api.status.{DependencyWarning, TaskRunReason}
 import za.co.absa.pramen.api.{Reason, Transformer}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.Metastore
@@ -40,7 +40,7 @@ class TransformationJob(operationDef: OperationDef,
 
   override val scheduleStrategy: ScheduleStrategy = new ScheduleStrategySourcing
 
-  override def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
+  override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
     preRunTransformationCheck(infoDate, dependencyWarnings)
   }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
index 565ff4c51..de2683a82 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
@@ -119,7 +119,7 @@ abstract class TaskRunnerBase(conf: Config,
   protected def runTask(task: Task): RunStatus = {
     val started = Instant.now()
     task.job.operation.killMaxExecutionTimeSeconds match {
-      case Some(timeout) =>
+      case Some(timeout) if timeout > 0 =>
         @volatile var runStatus: RunStatus = null
 
         try {
@@ -132,6 +132,9 @@ abstract class TaskRunnerBase(conf: Config,
           case NonFatal(ex) =>
             failTask(task, started, ex)
         }
+      case Some(timeout) =>
+        log.error(s"Incorrect timeout for the task: ${task.job.name}. Should be bigger than zero, got: $timeout.")
+        doValidateAndRunTask(task)
       case None =>
         doValidateAndRunTask(task)
     }
@@ -216,7 +219,7 @@ abstract class TaskRunnerBase(conf: Config,
     val isRawFileBased = task.job.outputTable.format.isInstanceOf[DataFormat.Raw]
 
     Try {
-      task.job.preRunCheck(task.infoDate, conf)
+      task.job.preRunCheck(task.infoDate, task.reason, conf)
     } match {
       case Success(validationResult) =>
         val resultToReturn = validationResult.status match {
@@ -262,8 +265,8 @@ abstract class TaskRunnerBase(conf: Config,
     * If validation is successful, in instance of JobPreRunResult is returned.
     * If validation failed an instance of TaskResult is returned.
     *
-    * @param task    a task to validate.
-    * @param started the instant when the job has started executing.
+    * @param task      a task to validate.
+    * @param started   the instant when the job has started executing.
     * @return an instance of TaskResult on validation failure or optional record count on success.
     */
   private[core] def validate(task: Task, started: Instant): Either[TaskResult, JobPreRunResult] = {
@@ -445,12 +448,42 @@ abstract class TaskRunnerBase(conf: Config,
     val updatedResult = taskResult.copy(notificationTargetErrors = notificationTargetErrors)
 
     logTaskResult(updatedResult, isLazy)
-    pipelineState.addTaskCompletion(Seq(updatedResult))
-    addJournalEntry(task, updatedResult, pipelineState.getState().pipelineInfo)
+    val wasInterrupted = isTaskInterrupted(task, taskResult)
+    if (wasInterrupted) {
+      log.warn("Skipping the interrupted exception of the killed task.")
+    } else {
+      pipelineState.addTaskCompletion(Seq(updatedResult))
+      addJournalEntry(task, updatedResult, pipelineState.getState().pipelineInfo)
+    }
 
     updatedResult.runStatus
   }
 
+  private def isTaskInterrupted(task: Task, taskResult: TaskResult): Boolean = {
+    val hasTimeout = task.job.operation.killMaxExecutionTimeSeconds.nonEmpty
+
+    taskResult.runStatus match {
+      case _: RunStatus.Failed if hasTimeout =>
+        val failureException = taskResult.runStatus.asInstanceOf[RunStatus.Failed].ex
+
+        failureException match {
+          case _: InterruptedException =>
+            true
+          case _: FatalErrorWrapper =>
+            failureException.getCause match {
+              case _: InterruptedException =>
+                true
+              case _ =>
+                false
+            }
+          case _ =>
+            false
+        }
+      case _ =>
+        false
+    }
+  }
+
   private def addJournalEntry(task: Task, taskResult: TaskResult, pipelineInfo: PipelineInfo): Unit = {
     val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult, pipelineInfo)
 


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala
index 331dedaa9..e19cf6acc 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobBaseDummy.scala
@@ -19,7 +19,7 @@ package za.co.absa.pramen.core.mocks.job
 import com.typesafe.config.Config
 import org.apache.spark.sql.DataFrame
 import za.co.absa.pramen.api.Reason
-import za.co.absa.pramen.api.status.DependencyWarning
+import za.co.absa.pramen.api.status.{DependencyWarning, TaskRunReason}
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.metastore.Metastore
 import za.co.absa.pramen.core.metastore.model.MetaTable
@@ -35,7 +35,7 @@ class JobBaseDummy(operationDef: OperationDef,
                    outputTableDef: MetaTable)
   extends JobBase(operationDef, metastore, bookkeeper, jobNotificationTargets, outputTableDef) {
 
-  override def preRunCheckJob(infoDate: LocalDate, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
+  override def preRunCheckJob(infoDate: LocalDate, runReason: TaskRunReason, jobConfig: Config, dependencyWarnings: Seq[DependencyWarning]): JobPreRunResult = {
     JobPreRunResult(null, None, dependencyWarnings, Seq.empty[String])
   }
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobSpy.scala
index 89fff6344..dd7599496 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobSpy.scala
@@ -19,6 +19,7 @@ package za.co.absa.pramen.core.mocks.job
 import com.typesafe.config.Config
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.types.StructType
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.api.{DataFormat, Reason}
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.metastore.MetaTableStats
@@ -67,6 +68,7 @@ class JobSpy(jobName: String = "DummyJob",
   override def trackDays: Int = jobTrackDays
 
   def preRunCheck(infoDate: LocalDate,
+                  runReason: TaskRunReason,
                   conf: Config): JobPreRunResult = {
     preRunCheckCount += 1
     preRunCheckFunction()
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala
index a21870867..582884163 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/metastore/MetastoreSpy.scala
@@ -38,6 +38,7 @@ class MetastoreSpy(registeredTables: Seq[String] = Seq("table1", "table2"),
                    isTableAvailable: Boolean = true,
                    isTableEmpty: Boolean = false,
                    trackDays: Int = 0,
+                   failHive: Boolean = false,
                    readOptions: Map[String, String] = Map.empty[String, String],
                    writeOptions: Map[String, String] = Map.empty[String, String]) extends Metastore {
 
@@ -86,7 +87,10 @@ class MetastoreSpy(registeredTables: Seq[String] = Seq("table1", "table2"),
                                        schema: Option[StructType],
                                        hiveHelper: HiveHelper,
                                        recreate: Boolean): Unit = {
-    hiveCreationInvocations.append((tableName, infoDate, schema, recreate))
+    if (failHive) {
+      throw new RuntimeException("Test exception")
+    } else
+      hiveCreationInvocations.append((tableName, infoDate, schema, recreate))
   }
 
   override def getStats(tableName: String, infoDate: LocalDate): MetaTableStats = {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/IngestionJobSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/IngestionJobSuite.scala
index 6af247553..080bd80b3 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/IngestionJobSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/IngestionJobSuite.scala
@@ -19,6 +19,7 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}
 import org.apache.spark.sql.DataFrame
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.api.{Query, Reason}
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
@@ -43,6 +44,8 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
   private def exampleDf: DataFrame = List(("A", 1), ("B", 2), ("C", 3)).toDF("a", "b")
 
+  private val runReason: TaskRunReason = TaskRunReason.New
+
   private val conf: Config = ConfigFactory.parseString(
     s"""
        | pramen {
@@ -120,7 +123,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
         bk.setRecordCount("table1", infoDate, infoDate, infoDate, 4, 4, 123, 456, isTableTransient = false)
 
-        val result = job.preRunCheckJob(infoDate, conf, Nil)
+        val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
         assert(result.status == JobPreRunStatus.AlreadyRan)
       }
@@ -131,7 +134,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
           bk.setRecordCount("table1", infoDate, infoDate, infoDate, 100, 100, 123, 456, isTableTransient = false)
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.NeedsUpdate)
         }
@@ -141,7 +144,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
           bk.setRecordCount("table1", infoDate, infoDate, infoDate, 100, 100, 123, 456, isTableTransient = false)
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.NeedsUpdate)
         }
@@ -151,7 +154,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
         "some records, default minimum records" in {
           val (_, _, job) = getUseCase()
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.Ready)
         }
@@ -159,7 +162,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
         "some records, custom minimum records" in {
           val (_, _, job) = getUseCase(minRecords = Some(3))
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.Ready)
         }
@@ -167,7 +170,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
         "empty table, custom zero minimum records" in {
           val (_, _, job) = getUseCase(minRecords = Some(0), sourceTable = "empty")
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.Ready)
         }
@@ -177,7 +180,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
         "no records, default minimum records" in {
           val (_, _, job) = getUseCase(sourceTable = "empty")
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.NoData(false))
         }
@@ -187,7 +190,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
         "new job, some records, custom minimum records" in {
           val (_, _, job) = getUseCase(minRecords = Some(5))
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.InsufficientData(4, 5, None))
         }
@@ -197,7 +200,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
           bk.setRecordCount("table1", infoDate, infoDate, infoDate, 100, 100, 123, 456, isTableTransient = false)
 
-          val result = job.preRunCheckJob(infoDate, conf, Nil)
+          val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
           assert(result.status == JobPreRunStatus.InsufficientData(4, 5, Some(100)))
         }
@@ -212,7 +215,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
         bk.setRecordCount("table1", noDataInfoDate, noDataInfoDate, noDataInfoDate, 4, 4, 123, 456, isTableTransient = false)
 
-        val result = job.preRunCheckJob(noDataInfoDate, conf, Nil)
+        val result = job.preRunCheckJob(noDataInfoDate, runReason, conf, Nil)
 
         assert(result.status == JobPreRunStatus.AlreadyRan)
       }
@@ -224,7 +227,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
         bk.setRecordCount("table1", noDataInfoDate, noDataInfoDate, noDataInfoDate, 30, 30, 123, 456, isTableTransient = false)
 
-        val result = job.preRunCheckJob(noDataInfoDate, conf, Nil)
+        val result = job.preRunCheckJob(noDataInfoDate, runReason, conf, Nil)
 
         assert(result.status == JobPreRunStatus.NeedsUpdate)
       }
@@ -234,7 +237,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
         val noDataInfoDate = infoDate.plusDays(1)
 
-        val result = job.preRunCheckJob(noDataInfoDate, conf, Nil)
+        val result = job.preRunCheckJob(noDataInfoDate, runReason, conf, Nil)
 
         assert(result.status == JobPreRunStatus.Ready)
       }
@@ -244,7 +247,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
         val noDataInfoDate = infoDate.plusDays(3)
 
-        val result = job.preRunCheckJob(noDataInfoDate, conf, Nil)
+        val result = job.preRunCheckJob(noDataInfoDate, runReason, conf, Nil)
 
         assert(result.status == JobPreRunStatus.NoData(false))
       }
@@ -283,7 +286,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
 
         val (_, _, job) = getUseCase(tempDirectory = Option(tempDir), disableCountQuery = true)
 
-        val preRunCheck = job.preRunCheckJob(infoDate, conf, Seq.empty)
+        val preRunCheck = job.preRunCheckJob(infoDate, runReason, conf, Seq.empty)
         assert(preRunCheck.status == JobPreRunStatus.Ready)
         assert(preRunCheck.inputRecordsCount.contains(4))
         assert(TransientTableManager.hasDataForTheDate("source_cache://jdbc|company|2022-02-18", infoDate))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala
index 6fafd09eb..f00ebd20c 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/JobBaseSuite.scala
@@ -19,11 +19,12 @@ package za.co.absa.pramen.core.pipeline
 import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.DataFrame
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.api.status.MetastoreDependency
+import za.co.absa.pramen.api.status.{MetastoreDependency, TaskRunReason}
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.expr.exceptions.SyntaxErrorException
 import za.co.absa.pramen.core.fixtures.TextComparisonFixture
+import za.co.absa.pramen.core.metastore.model.HiveConfig
 import za.co.absa.pramen.core.mocks.MetaTableFactory
 import za.co.absa.pramen.core.mocks.bookkeeper.SyncBookkeeperMock
 import za.co.absa.pramen.core.mocks.job.JobBaseDummy
@@ -35,6 +36,7 @@ import java.time.{Instant, LocalDate}
 class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFixture {
 
   private val infoDate = LocalDate.of(2022, 1, 18)
+  private val runReason: TaskRunReason = TaskRunReason.New
 
   "allowRunningTasksInParallel()" should {
     "be true for jobs that don't have dependencies" in {
@@ -129,7 +131,7 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
       val dep = MetastoreDependency(Seq("table1"), "@infoDate", None, triggerUpdates = false, isOptional = false, isPassive = false)
       val job = getUseCase(dependencies = Seq(dep), isTableAvailable = false)
 
-      val actual = job.preRunCheck(infoDate, conf)
+      val actual = job.preRunCheck(infoDate, runReason, conf)
 
       assert(actual.dependencyWarnings.isEmpty)
       assert(actual.status.isInstanceOf[FailedDependencies])
@@ -143,7 +145,7 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
       val dep = MetastoreDependency(Seq("table1"), "@infoDate", None, triggerUpdates = false, isOptional = false, isPassive = true)
       val job = getUseCase(dependencies = Seq(dep), isTableAvailable = false)
 
-      val actual = job.preRunCheck(infoDate, conf)
+      val actual = job.preRunCheck(infoDate, runReason, conf)
 
       assert(actual.dependencyWarnings.isEmpty)
       assert(actual.status.isInstanceOf[FailedDependencies])
@@ -157,7 +159,7 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
       val dep = MetastoreDependency(Seq("table2"), "@infoDate", None, triggerUpdates = false, isOptional = false, isPassive = false)
       val job = getUseCase(dependencies = Seq(dep), isTableAvailable = false, isTableEmpty = true)
 
-      val actual = job.preRunCheck(infoDate, conf)
+      val actual = job.preRunCheck(infoDate, runReason, conf)
 
       assert(actual.dependencyWarnings.isEmpty)
       assert(actual.status.isInstanceOf[FailedDependencies])
@@ -169,7 +171,7 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
       val dep = MetastoreDependency(Seq("table1"), "@infoDate", None, triggerUpdates = false, isOptional = true, isPassive = false)
       val job = getUseCase(dependencies = Seq(dep), isTableAvailable = false)
 
-      val actual = job.preRunCheck(infoDate.plusDays(1), conf)
+      val actual = job.preRunCheck(infoDate.plusDays(1), runReason, conf)
 
       assert(actual.dependencyWarnings.nonEmpty)
       assert(actual.dependencyWarnings.head.table == "table1")
@@ -211,12 +213,53 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
     }
   }
 
+  "createOrRefreshHiveTable" should {
+    "do nothing if Hive table is not defined" in {
+      val job = getUseCase()
+
+      val warnings = job.createOrRefreshHiveTable(null, infoDate, recreate = false)
+
+      assert(warnings.isEmpty)
+
+    }
+
+    "return an empty seq of warnings when the operation succeeded" in {
+      val job = getUseCase(hiveTable = Some("test_hive_table"))
+
+      val warnings = job.createOrRefreshHiveTable(null, infoDate, recreate = false)
+
+      assert(warnings.isEmpty)
+    }
+
+    "return warnings if ignore failures enabled" in {
+      val job = getUseCase(hiveTable = Some("test_hive_table"), hiveFailure = true, ignoreHiveFailures = true)
+
+      val warnings = job.createOrRefreshHiveTable(null, infoDate, recreate = false)
+
+      assert(warnings.nonEmpty)
+      assert(warnings.head == "Failed to create or update Hive table 'test_hive_table': Test exception")
+    }
+
+    "re-throw the exception if ignore failures disabled" in {
+      val job = getUseCase(hiveTable = Some("test_hive_table"), hiveFailure = true)
+
+      val ex = intercept[RuntimeException] {
+        job.createOrRefreshHiveTable(null, infoDate, recreate = false)
+      }
+
+      assert(ex.getMessage == "Test exception")
+    }
+  }
+
   def getUseCase(tableDf: DataFrame = null,
                  dependencies: Seq[MetastoreDependency] = Nil,
                  isTableAvailable: Boolean = true,
                  isTableEmpty: Boolean = false,
                  allowParallel: Boolean = true,
-                 warnMaxExecutionTimeSeconds: Option[Int] = None): JobBase = {
+                 warnMaxExecutionTimeSeconds: Option[Int] = None,
+                 hiveTable: Option[String] = None,
+                 hiveFailure: Boolean = false,
+                 ignoreHiveFailures: Boolean = false): JobBase = {
     val operation = OperationDefFactory.getDummyOperationDef(dependencies = dependencies,
       allowParallel = allowParallel,
       warnMaxExecutionTimeSeconds = warnMaxExecutionTimeSeconds,
@@ -224,9 +267,12 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
 
     val bk = new SyncBookkeeperMock
 
-    val metastore = new MetastoreSpy(tableDf = tableDf, isTableAvailable = isTableAvailable, isTableEmpty = isTableEmpty)
+    val metastore = new MetastoreSpy(tableDf = tableDf, isTableAvailable = isTableAvailable, isTableEmpty = isTableEmpty, failHive = hiveFailure)
 
-    val outputTable = MetaTableFactory.getDummyMetaTable(name = "test_output_table")
+    val outputTable = MetaTableFactory.getDummyMetaTable(name = "test_output_table",
+      hiveTable = hiveTable,
+      hiveConfig = HiveConfig.getNullConfig.copy(ignoreFailures = ignoreHiveFailures)
+    )
 
     new JobBaseDummy(operation, Nil, metastore, bk, outputTable)
   }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJobSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJobSuite.scala
index 47ffd8d08..e268f076b 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJobSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/PythonTransformationJobSuite.scala
@@ -22,6 +22,7 @@ import org.apache.spark.sql.catalyst.analysis.NoSuchDatabaseException
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.Reason
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.exceptions.ProcessFailedException
@@ -43,6 +44,7 @@ class PythonTransformationJobSuite extends AnyWordSpec with BeforeAndAfterAll wi
   private val infoDate = LocalDate.of(2022, 2, 18)
   private var tempDir: String = _
   private val exampleDf: DataFrame = List(("A", 1), ("B", 2), ("C", 3)).toDF("a", "b")
+  private val runReason: TaskRunReason = TaskRunReason.New
 
   private def getConf(tempDir: String): Config = ConfigFactory.parseString(
     s"""
@@ -120,7 +122,7 @@ class PythonTransformationJobSuite extends AnyWordSpec with BeforeAndAfterAll wi
     "always return Ready" in {
       val (job, _, _, _) = getUseCase()
 
-      val result = job.preRunCheckJob(infoDate, conf, Nil)
+      val result = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(result.status == JobPreRunStatus.Ready)
     }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkJobSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkJobSuite.scala
index 403bded8b..8ad30223d 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkJobSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkJobSuite.scala
@@ -20,6 +20,7 @@ import com.typesafe.config.{ConfigFactory, ConfigValueFactory}
 import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.{AnalysisException, DataFrame}
 import org.scalatest.wordspec.AnyWordSpec
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.api.{Reason, Sink}
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
@@ -38,6 +39,7 @@ class SinkJobSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
 
   private val infoDate = LocalDate.of(2022, 1, 18)
   private val conf = ConfigFactory.empty()
+  private val runReason: TaskRunReason = TaskRunReason.New
 
   private def exampleDf: DataFrame = List(("A", 1), ("B", 2), ("C", 3)).toDF("a", "b")
 
@@ -45,7 +47,7 @@ class SinkJobSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
     "return Ready when the input table is available" in {
       val (job, _) = getUseCase(tableDf = exampleDf)
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.Ready, Some(3), Nil, Nil))
     }
@@ -53,7 +55,7 @@ class SinkJobSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
     "return Ready when the input table has no data" in {
       val (job, _) = getUseCase(tableDf = exampleDf.filter(col("a") > 3))
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.Ready, Some(0), Nil, Nil))
     }
@@ -62,7 +64,7 @@ class SinkJobSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
       val (job, _) = getUseCase(tableException = new RuntimeException("Dummy Exception"))
 
       val ex = intercept[IllegalStateException] {
-        job.preRunCheckJob(infoDate, conf, Nil)
+        job.preRunCheckJob(infoDate, runReason, conf, Nil)
       }
 
       assert(ex.getMessage == "Unable to read input table table1 for 2022-01-18.")
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferJobSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferJobSuite.scala
index c2b86aa0d..0acd9e204 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferJobSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferJobSuite.scala
@@ -20,6 +20,7 @@ import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.{AnalysisException, DataFrame}
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.Reason
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.fixtures.{TempDirFixture, TextComparisonFixture}
@@ -37,6 +38,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
 
   private val infoDate = LocalDate.of(2022, 1, 18)
   private val conf = ConfigFactory.empty()
+  private val runReason: TaskRunReason = TaskRunReason.New
 
   private def exampleDf: DataFrame = List(("A", 1), ("B", 2), ("C", 3)).toDF("a", "b")
 
@@ -44,7 +46,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
     "return Ready when the input table is available" in {
       val (job, _) = getUseCase()
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.Ready, Some(5), Nil, Nil))
     }
@@ -52,7 +54,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
     "return NoData when the input table has no data" in {
       val (job, _) = getUseCase(numberOfRecords = 0)
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.NoData(false), None, Nil, Nil))
     }
@@ -60,7 +62,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
     "return NoData with a failure when the input table has no data" in {
       val (job, _) = getUseCase(numberOfRecords = 0, failOnNoData = true)
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.NoData(true), None, Nil, Nil))
     }
@@ -71,7 +73,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
 
       bk.setRecordCount("table1->sink", infoDate, infoDate, infoDate, 3, 3, 10000, 1001, isTableTransient = false)
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.NeedsUpdate, Some(7), Nil, Nil))
     }
@@ -81,7 +83,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
 
       bk.setRecordCount("table1->sink", infoDate, infoDate, infoDate, 7, 3, 10000, 1001, isTableTransient = false)
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual == JobPreRunResult(JobPreRunStatus.AlreadyRan, Some(7), Nil, Nil))
     }
@@ -90,7 +92,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
       val (job, _) = getUseCase(getCountException = new RuntimeException("Dummy Exception"))
 
       val ex = intercept[RuntimeException] {
-        job.preRunCheckJob(infoDate, conf, Nil)
+        job.preRunCheckJob(infoDate, runReason, conf, Nil)
       }
 
       assert(ex.getMessage == "Dummy Exception")
@@ -136,7 +138,7 @@ class TransferJobSuite extends AnyWordSpec with SparkTestBase with TextCompariso
       withTempDirectory("cached_transfer_data") { tempDir =>
         val (job, _) = getUseCase(tempDirectory = Option(tempDir), disableCountQuery = true)
 
-        val preRunCheck = job.preRunCheckJob(infoDate, conf, Seq.empty)
+        val preRunCheck = job.preRunCheckJob(infoDate, runReason, conf, Seq.empty)
         assert(preRunCheck.status == JobPreRunStatus.Ready)
         assert(preRunCheck.inputRecordsCount.contains(5))
         assert(TransientTableManager.hasDataForTheDate("source_cache://testsource|table1|2022-01-18", infoDate))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransformationJobSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransformationJobSuite.scala
index ce78355f6..7b9cb1f5f 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransformationJobSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransformationJobSuite.scala
@@ -20,6 +20,7 @@ import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.DataFrame
 import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.pramen.api.Reason
+import za.co.absa.pramen.api.status.TaskRunReason
 import za.co.absa.pramen.core.OperationDefFactory
 import za.co.absa.pramen.core.base.SparkTestBase
 import za.co.absa.pramen.core.mocks.MetaTableFactory
@@ -33,6 +34,7 @@ class TransformationJobSuite extends AnyWordSpec with SparkTestBase {
 
   private val infoDate = LocalDate.of(2022, 1, 18)
   private val conf = ConfigFactory.empty()
+  private val runReason: TaskRunReason = TaskRunReason.New
 
   private def exampleDf: DataFrame = List(("A", 1), ("B", 2), ("C", 3)).toDF("a", "b")
 
@@ -40,7 +42,7 @@ class TransformationJobSuite extends AnyWordSpec with SparkTestBase {
     "return Ready" in {
       val (job, _) = getUseCase(validateFunction = () => Reason.Ready)
 
-      val actual = job.preRunCheckJob(infoDate, conf, Nil)
+      val actual = job.preRunCheckJob(infoDate, runReason, conf, Nil)
 
       assert(actual.status == JobPreRunStatus.Ready)
     }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
index abaac2555..7a918de9f 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
@@ -47,6 +47,7 @@ import scala.concurrent.Await
 import scala.concurrent.duration.Duration
 
 class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFixture {
+
   import spark.implicits._
 
   private val infoDate = LocalDate.of(2022, 2, 18)
@@ -145,6 +146,43 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
       assert(journalEntries.head.status == "Failed")
     }
 
+    "run a job that is failing with timeout" in {
+      val now = Instant.now()
+
+      val runFunction: () => RunResult = () => {
+        Thread.sleep(2000)
+        null
+      }
+
+      val (runner, _, journal, state, tasks) = getUseCase(runFunction = runFunction,
+        isRerun = true,
+        allowParallel = false,
+        timeoutTask = true)
+
+      val taskPreDefs = Seq(core.pipeline.TaskPreDef(infoDate, TaskRunReason.New))
+
+      val fut = runner.runJobTasks(tasks.head.job, taskPreDefs)
+
+      Await.result(fut, Duration.Inf)
+
+      val result = state.completedStatuses
+
+      val job = tasks.head.job.asInstanceOf[JobSpy]
+
+      assert(job.validateCount == 1)
+      assert(job.runCount == 1)
+      assert(job.postProcessingCount == 0)
+      assert(job.saveCount == 0)
+      assert(job.createHiveTableCount == 0)
+      assert(result.length == 1)
+      assert(result.head.runStatus.isInstanceOf[Failed])
+
+      val journalEntries = journal.getEntries(now, now.plusSeconds(30))
+
+      assert(journalEntries.length == 1)
+      assert(journalEntries.head.status == "Failed")
+    }
+
     "run a single lazy job" in {
       val now = Instant.now()
       val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskResult) => ())
@@ -581,7 +619,8 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
                  bookkeeperIn: Bookkeeper = null,
                  allowParallel: Boolean = true,
                  hiveTable: Option[String] = None,
-                 jobNotificationTargets: Seq[JobNotificationTarget] = Nil
+                 jobNotificationTargets: Seq[JobNotificationTarget] = Nil,
+                 timeoutTask: Boolean = false
                 ): (TaskRunnerBase, Bookkeeper, Journal, PipelineStateSpy, Seq[Task]) = {
     val conf = ConfigFactory.empty()
 
@@ -590,12 +629,13 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
     val bookkeeper = if (bookkeeperIn == null) new SyncBookkeeperMock else bookkeeperIn
     val journal = new JournalMock
     val tokenLockFactory = new TokenLockFactoryMock
-
     val state = new PipelineStateSpy
+    val killTimer = if (timeoutTask) Some(1) else None
 
     val operationDef = OperationDefFactory.getDummyOperationDef(
       schemaTransformations = List(TransformExpression("c", Some("cast(b as string)"), None)),
-      filters = List("b > 1")
+      filters = List("b > 1"),
+      killMaxExecutionTimeSeconds = killTimer
     )
 
     val stats = MetaTableStats(2, Some(100))