Repository: AbsaOSS/pramen (language: python)

Issue:
Specify maximum time for a job and warn if job took longer
## Background
This is so if a job takes too long, it can be easier identifiable in notifications.

## Feature
Specify maximum time for a job and warn if job took longer.

## Example [Optional]
--

## Proposed Solution [Optional]
--


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index bb57795b0..2c5c61c7e 100644
--- a/README.md
+++ b/README.md
@@ -1919,7 +1919,11 @@ Here is an example configuration for a JDBC source:
       # You can define range queries to the input table by providing date expressions like this:
       date.from = "@infoDate - 1"
       date.to = "@infoDate"
-      
+
+      # [Optional] You can specify the maximum about the job should take. If the execution time is bigger than
+      # specified, a warning will be added to notifications.
+      warn.maximum.execution.time.seconds = 3600
+
       # You can override any of source settings here 
       source {
         minimum.records = 1000 
@@ -2048,6 +2052,10 @@ Here is a example:
     key2 = "value2"
     key3 = ${MY_ENV_VARIABLE}
   }
+
+  # [Optional] You can specify the maximum about the job should take. If the execution time is bigger than
+  # specified, a warning will be added to notifications.
+  warn.maximum.execution.time.seconds = 3600
   
   # Optional schema transformations 
   transformations = [
@@ -2175,6 +2183,10 @@ sink operation definition.
         from = "lastMonday(@infoDate) - 7"
         to = "lastSunday(@infoDate)"
       }
+
+      # [Optional] You can specify the maximum about the job should take. If the execution time is bigger than
+      # specified, a warning will be added to notifications.
+      warn.maximum.execution.time.seconds = 3600
     }
   ]
 }
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 89108e669..f73509435 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
@@ -170,7 +170,10 @@ class IngestionJob(operationDef: OperationDef,
 
     source.close()
 
-    SaveResult(stats)
+    val jobFinished = Instant.now
+    val tooLongWarnings = getTookTooLongWarnings(jobStarted, jobFinished, sourceTable.warnMaxExecutionTimeSeconds)
+
+    SaveResult(stats, warnings = tooLongWarnings)
   }
 
   private def supportsTrackDays(hasInfoDate: Boolean): Boolean = {
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 cfebc5f75..a70fbbb7c 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
@@ -24,10 +24,10 @@ import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.expr.DateExprEvaluator
 import za.co.absa.pramen.core.metastore.Metastore
 import za.co.absa.pramen.core.metastore.model.MetaTable
-import za.co.absa.pramen.core.pipeline
 import za.co.absa.pramen.core.utils.Emoji._
+import za.co.absa.pramen.core.utils.TimeUtils
 
-import java.time.LocalDate
+import java.time.{Instant, LocalDate}
 import scala.util.{Failure, Success, Try}
 
 abstract class JobBase(operationDef: OperationDef,
@@ -198,6 +198,27 @@ abstract class JobBase(operationDef: OperationDef,
 
     (effectiveFrom, effectiveTo)
   }
+
+  private[core] def getTookTooLongWarnings(jobStarted: Instant, jobFinished: Instant, maxTimeSecondsOpt: Option[Int]): Seq[String] = {
+    val effectiveMaxTimeOpt = maxTimeSecondsOpt match {
+      case Some(maxTimeTable) =>
+        operationDef.warnMaxExecutionTimeSeconds match {
+          case Some(maxTimeOp) => Option(Math.min(maxTimeTable, maxTimeOp))
+          case None => maxTimeSecondsOpt
+        }
+      case None => operationDef.warnMaxExecutionTimeSeconds
+    }
+    effectiveMaxTimeOpt.flatMap { maxTimeSeconds =>
+      val actualTime = jobFinished.getEpochSecond - jobStarted.getEpochSecond
+      if (actualTime > maxTimeSeconds) {
+        val prettyActualTime = TimeUtils.prettyPrintElapsedTimeShort(actualTime * 1000)
+        val prettyMaxTime = TimeUtils.prettyPrintElapsedTimeShort(maxTimeSeconds * 1000)
+        Some(s"The job took longer ($prettyActualTime) than expected ($prettyMaxTime).")
+      } else {
+        None
+      }
+    }.toSeq
+  }
 }
 
 object JobBase {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
index 48e1a992a..ae43bd452 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
@@ -41,6 +41,7 @@ case class OperationDef(
                          outputInfoDateExpression: String,
                          initialSourcingDateExpression: String,
                          processingTimestampColumn: Option[String],
+                         warnMaxExecutionTimeSeconds: Option[Int],
                          schemaTransformations: Seq[TransformExpression],
                          filters: Seq[String],
                          notificationTargets: Seq[String],
@@ -63,6 +64,7 @@ object OperationDef {
   val OUTPUT_INFO_DATE_EXPRESSION_KEY = "info.date.expr"
   val INITIAL_SOURCING_DATE_EXPR = "initial.sourcing.date.expr"
   val PROCESSING_TIMESTAMP_COLUMN_KEY = "processing.timestamp.column"
+  val WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY = "warn.maximum.execution.time.seconds"
   val SCHEMA_TRANSFORMATIONS_KEY = "transformations"
   val FILTERS_KEY = "filters"
   val NOTIFICATION_TARGETS_KEY = "notification.targets"
@@ -92,6 +94,7 @@ object OperationDef {
     val outputInfoDateExpressionOpt = ConfigUtils.getOptionString(conf, OUTPUT_INFO_DATE_EXPRESSION_KEY)
     val initialSourcingDateExpressionOpt = ConfigUtils.getOptionString(conf, INITIAL_SOURCING_DATE_EXPR)
     val processingTimestampColumn = ConfigUtils.getOptionString(conf, PROCESSING_TIMESTAMP_COLUMN_KEY)
+    val maximumExecutionTimeSeconds = ConfigUtils.getOptionInt(conf, WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY)
     val schemaTransformations = TransformExpression.fromConfig(conf, SCHEMA_TRANSFORMATIONS_KEY, parent)
     val filters = ConfigUtils.getOptListStrings(conf, FILTERS_KEY)
     val notificationTargets = ConfigUtils.getOptListStrings(conf, NOTIFICATION_TARGETS_KEY)
@@ -130,6 +133,7 @@ object OperationDef {
       outputInfoDateExpression,
       initialSourcingDateExpression,
       processingTimestampColumn,
+      maximumExecutionTimeSeconds,
       schemaTransformations,
       filters,
       notificationTargets,
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 3a691983c..f904812bf 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
@@ -128,6 +128,8 @@ class SinkJob(operationDef: OperationDef,
 
       val isTransient = outputTable.format.isTransient
 
+      val tooLongWarnings = getTookTooLongWarnings(jobStarted, jobFinished, sinkTable.warnMaxExecutionTimeSeconds)
+
       bookkeeper.setRecordCount(outputTable.name,
         infoDate,
         infoDate,
@@ -140,7 +142,7 @@ class SinkJob(operationDef: OperationDef,
       )
 
       val stats = MetaTableStats(sinkResult.recordsSent, None)
-      SaveResult(stats, sinkResult.filesSent, sinkResult.hiveTables, sinkResult.warnings)
+      SaveResult(stats, sinkResult.filesSent, sinkResult.hiveTables, sinkResult.warnings ++ tooLongWarnings)
     } catch {
       case NonFatal(ex) => throw new IllegalStateException("Unable to write to the sink.", ex)
     } finally {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkTable.scala
index f89c1972a..16399addb 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SinkTable.scala
@@ -18,6 +18,7 @@ package za.co.absa.pramen.core.pipeline
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
+import za.co.absa.pramen.core.pipeline.OperationDef.WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY
 import za.co.absa.pramen.core.utils.{AlgorithmUtils, ConfigUtils}
 
 import scala.collection.JavaConverters._
@@ -28,6 +29,7 @@ case class SinkTable(
                       conf: Config,
                       rangeFromExpr: Option[String],
                       rangeToExpr: Option[String],
+                      warnMaxExecutionTimeSeconds: Option[Int],
                       transformations: Seq[TransformExpression],
                       filters: Seq[String],
                       columns: Seq[String],
@@ -56,6 +58,7 @@ object SinkTable {
     val outputTableName = ConfigUtils.getOptionString(conf, JOB_METASTORE_OUTPUT_TABLE_KEY)
     val dateFromExpr = ConfigUtils.getOptionString(conf, DATE_FROM_KEY)
     val dateToExpr = ConfigUtils.getOptionString(conf, DATE_TO_KEY)
+    val maximumExecutionTimeSeconds = ConfigUtils.getOptionInt(conf, WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY)
     val transformations = TransformExpression.fromConfig(conf, TRANSFORMATIONS_KEY, parentPath)
     val filters = ConfigUtils.getOptListStrings(conf, FILTERS_KEY)
     val columns = ConfigUtils.getOptListStrings(conf, COLUMNS_KEY)
@@ -68,7 +71,7 @@ object SinkTable {
       None
     }
 
-    SinkTable(metaTableName, outputTableName, conf, dateFromExpr, dateToExpr, transformations, filters, columns, options, overrideConf)
+    SinkTable(metaTableName, outputTableName, conf, dateFromExpr, dateToExpr, maximumExecutionTimeSeconds, transformations, filters, columns, options, overrideConf)
   }
 
   def fromConfig(conf: Config, arrayPath: String): Seq[SinkTable] = {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SourceTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SourceTable.scala
index daa4af566..d94d17b79 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SourceTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/SourceTable.scala
@@ -20,6 +20,7 @@ import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
 import za.co.absa.pramen.api.Query
 import za.co.absa.pramen.core.model.QueryBuilder
+import za.co.absa.pramen.core.pipeline.OperationDef.WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY
 import za.co.absa.pramen.core.utils.{AlgorithmUtils, ConfigUtils}
 
 import scala.collection.JavaConverters._
@@ -30,6 +31,7 @@ case class SourceTable(
                         conf: Config,
                         rangeFromExpr: Option[String],
                         rangeToExpr: Option[String],
+                        warnMaxExecutionTimeSeconds: Option[Int],
                         transformations: Seq[TransformExpression],
                         filters: Seq[String],
                         columns: Seq[String],
@@ -56,6 +58,7 @@ object SourceTable {
     val query = QueryBuilder.fromConfig(conf, "input", parentPath)
     val dateFromExpr = ConfigUtils.getOptionString(conf, DATE_FROM_KEY)
     val dateToExpr = ConfigUtils.getOptionString(conf, DATE_TO_KEY)
+    val maximumExecutionTimeSeconds = ConfigUtils.getOptionInt(conf, WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY)
     val columns = ConfigUtils.getOptListStrings(conf, COLUMNS_KEY)
     val transformations = TransformExpression.fromConfig(conf, TRANSFORMATIONS_KEY, parentPath)
     val filters = ConfigUtils.getOptListStrings(conf, FILTERS_KEY)
@@ -67,7 +70,7 @@ object SourceTable {
       None
     }
 
-    SourceTable(metaTableName, query, conf, dateFromExpr, dateToExpr, transformations, filters, columns, overrideConf)
+    SourceTable(metaTableName, query, conf, dateFromExpr, dateToExpr, maximumExecutionTimeSeconds, transformations, filters, columns, overrideConf)
   }
 
   def fromConfig(conf: Config, arrayPath: String): Seq[SourceTable] = {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
index 5bda18cda..600048be2 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
@@ -23,6 +23,7 @@ import za.co.absa.pramen.core.app.config.InfoDateConfig
 import za.co.absa.pramen.core.config.InfoDateOverride
 import za.co.absa.pramen.core.metastore.model.{HiveConfig, MetaTable}
 import za.co.absa.pramen.core.model.QueryBuilder
+import za.co.absa.pramen.core.pipeline.OperationDef.WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY
 import za.co.absa.pramen.core.utils.{AlgorithmUtils, ConfigUtils}
 
 import java.time.LocalDate
@@ -37,6 +38,7 @@ case class TransferTable(
                           infoDateStart: LocalDate,
                           trackDays: Int,
                           trackDaysExplicitlySet: Boolean,
+                          warnMaxExecutionTimeSeconds: Option[Int],
                           transformations: Seq[TransformExpression],
                           filters: Seq[String],
                           columns: Seq[String],
@@ -46,11 +48,11 @@ case class TransferTable(
                           sinkOverrideConf: Option[Config]
                         ) {
   def getSourceTable: SourceTable = {
-    SourceTable(jobMetaTableName, query, conf, rangeFromExpr, rangeToExpr, transformations, filters, columns, sourceOverrideConf)
+    SourceTable(jobMetaTableName, query, conf, rangeFromExpr, rangeToExpr, warnMaxExecutionTimeSeconds, transformations, filters, columns, sourceOverrideConf)
   }
 
   def getSinkTable: SinkTable = {
-    SinkTable(jobMetaTableName, Option(jobMetaTableName), conf, rangeFromExpr, rangeToExpr, transformations, filters, columns, writeOptions, sinkOverrideConf)
+    SinkTable(jobMetaTableName, Option(jobMetaTableName), conf, rangeFromExpr, rangeToExpr, warnMaxExecutionTimeSeconds, transformations, filters, columns, writeOptions, sinkOverrideConf)
   }
 
   def getMetaTable: MetaTable = {
@@ -76,6 +78,7 @@ object TransferTable {
     val jobMetaTableOpt = ConfigUtils.getOptionString(conf, JOB_METASTORE_OUTPUT_TABLE_KEY)
     val dateFromExpr = ConfigUtils.getOptionString(conf, DATE_FROM_KEY)
     val dateToExpr = ConfigUtils.getOptionString(conf, DATE_TO_KEY)
+    val maximumExecutionTimeSeconds = ConfigUtils.getOptionInt(conf, WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY)
     val trackDays = ConfigUtils.getOptionInt(conf, TRACK_DAYS_KEY).getOrElse(defaultTrackDays)
     val trackDaysExplicitlySet = conf.hasPath(TRACK_DAYS_KEY)
     val columns = ConfigUtils.getOptListStrings(conf, COLUMNS_KEY)
@@ -104,7 +107,7 @@ object TransferTable {
     val startDate = infoDateOverride.startDate.getOrElse(defaultStartDate)
     val jobMetaTable = getOutputTableName(jobMetaTableOpt, query, sinkName)
 
-    TransferTable(query, jobMetaTable, conf, dateFromExpr, dateToExpr, startDate, trackDays, trackDaysExplicitlySet, transformations, filters, columns, readOptions, writeOptions, sourceOverrideConf, sinkOverrideConf)
+    TransferTable(query, jobMetaTable, conf, dateFromExpr, dateToExpr, startDate, trackDays, trackDaysExplicitlySet, maximumExecutionTimeSeconds, transformations, filters, columns, readOptions, writeOptions, sourceOverrideConf, sinkOverrideConf)
   }
 
   def fromConfig(conf: Config, infoDateConfig: InfoDateConfig, arrayPath: String, sinkName: String): Seq[TransferTable] = {
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 baad5c53a..78f7622f1 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
@@ -75,6 +75,9 @@ class TransformationJob(operationDef: OperationDef,
       case _: AbstractMethodError => log.warn(s"Transformers were built using old version of Pramen that does not support post processing. Ignoring...")
     }
 
-    saveResults
+    val jobFinished = Instant.now
+    val tooLongWarnings = getTookTooLongWarnings(jobStarted, jobFinished, None)
+
+    saveResults.copy(warnings = saveResults.warnings ++ tooLongWarnings)
   }
 }


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala
index 3b5f9def2..65caa01db 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/OperationDefFactory.scala
@@ -35,6 +35,7 @@ object OperationDefFactory {
                            outputInfoDateExpression: String = "@date",
                            initialSourcingDateExpression: String = "@runDate",
                            processingTimestampColumn: Option[String] = None,
+                           warnMaxExecutionTimeSeconds: Option[Int] = None,
                            schemaTransformations: Seq[TransformExpression] = Nil,
                            filters: Seq[String] = Nil,
                            notificationTargets: Seq[String] = Nil,
@@ -53,6 +54,7 @@ object OperationDefFactory {
       outputInfoDateExpression,
       initialSourcingDateExpression,
       processingTimestampColumn,
+      warnMaxExecutionTimeSeconds,
       schemaTransformations,
       filters,
       notificationTargets,
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SinkTableFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SinkTableFactory.scala
index cdfdc5ac2..d3864b4bc 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SinkTableFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SinkTableFactory.scala
@@ -25,11 +25,12 @@ object SinkTableFactory {
                         conf: Config = ConfigFactory.empty(),
                         sinkFromExpr: Option[String] = None,
                         sinkToExpr: Option[String] = None,
+                        warnMaxExecutionTimeSeconds: Option[Int] = None,
                         transformations: Seq[TransformExpression] = Nil,
                         filters: Seq[String] = Nil,
                         columns: Seq[String] = Nil,
                         options: Map[String, String] = Map.empty[String, String],
                         overrideConf: Option[Config] = None): SinkTable = {
-    SinkTable(metaTableName, outputTableName, conf, sinkFromExpr, sinkToExpr, transformations, filters, columns, options, overrideConf)
+    SinkTable(metaTableName, outputTableName, conf, sinkFromExpr, sinkToExpr, warnMaxExecutionTimeSeconds, transformations, filters, columns, options, overrideConf)
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SourceTableFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SourceTableFactory.scala
index 23576482d..f76eee9fa 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SourceTableFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/SourceTableFactory.scala
@@ -26,10 +26,11 @@ object SourceTableFactory {
                           conf: Config = ConfigFactory.empty(),
                           rangeFromExpr: Option[String] = None,
                           rangeToExpr: Option[String] = None,
+                          warnMaxExecutionTimeSeconds: Option[Int] = None,
                           transformations: Seq[TransformExpression] = Nil,
                           filters: Seq[String] = Nil,
                           columns: Seq[String] = Nil,
                           overrideConf: Option[Config] = None): SourceTable = {
-    SourceTable(metaTableName, query, conf, rangeFromExpr, rangeToExpr, transformations, filters, columns, overrideConf)
+    SourceTable(metaTableName, query, conf, rangeFromExpr, rangeToExpr, warnMaxExecutionTimeSeconds, transformations, filters, columns, overrideConf)
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala
index 1f6b26a8d..4199932f3 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala
@@ -31,6 +31,7 @@ object TransferTableFactory {
                             infoDateStart: LocalDate = LocalDate.of(2020, 1, 1),
                             trackDays: Int = 1,
                             trackDaysExplicitlySet: Boolean = false,
+                            warnMaxExecutionTimeSeconds: Option[Int] = None,
                             transformations: Seq[TransformExpression] = Nil,
                             filters: Seq[String] = Nil,
                             columns: Seq[String] = Nil,
@@ -46,6 +47,7 @@ object TransferTableFactory {
       infoDateStart,
       trackDays,
       trackDaysExplicitlySet,
+      warnMaxExecutionTimeSeconds,
       transformations,
       filters,
       columns,
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 2403eabcc..6af247553 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
@@ -402,7 +402,7 @@ class IngestionJobSuite extends AnyWordSpec with SparkTestBase with TextComparis
       Nil,
       sourceName,
       source,
-      SourceTable("table1", Query.Table(sourceTable), tableConf, rangeFromExpr, rangeToExpr, Seq(
+      SourceTable("table1", Query.Table(sourceTable), tableConf, rangeFromExpr, rangeToExpr, None, Seq(
         TransformExpression("NAME_U", Some("upper(NAME)"), None)
       ), Seq("ID > 1"), Seq("ID", "NAME", "NAME_U", "EMAIL"), configOverride),
       outputTable,
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 4e6b62efa..6fafd09eb 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
@@ -30,7 +30,7 @@ import za.co.absa.pramen.core.mocks.job.JobBaseDummy
 import za.co.absa.pramen.core.mocks.metastore.MetastoreSpy
 import za.co.absa.pramen.core.pipeline.JobPreRunStatus.FailedDependencies
 
-import java.time.LocalDate
+import java.time.{Instant, LocalDate}
 
 class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFixture {
 
@@ -176,13 +176,50 @@ class JobBaseSuite extends AnyWordSpec with SparkTestBase with TextComparisonFix
     }
   }
 
+  "getTookTooLongWarnings" should {
+    "return empty seq if maximum time is not defined" in {
+      val job = getUseCase()
+
+      val warnings = job.getTookTooLongWarnings(Instant.ofEpochSecond(1000), Instant.ofEpochSecond(1010), None)
+
+      assert(warnings.isEmpty)
+    }
+
+    "return empty seq if actual time is less than operation maximum time" in {
+      val job = getUseCase(warnMaxExecutionTimeSeconds = Some(100))
+
+      val warnings = job.getTookTooLongWarnings(Instant.ofEpochSecond(1000), Instant.ofEpochSecond(1010), None)
+
+      assert(warnings.isEmpty)
+    }
+
+    "return empty seq if actual time is less than table maximum time" in {
+      val job = getUseCase()
+
+      val warnings = job.getTookTooLongWarnings(Instant.ofEpochSecond(1000), Instant.ofEpochSecond(1010), Some(100))
+
+      assert(warnings.isEmpty)
+    }
+
+    "return a warning actual time is bigger than minimum of operation and table times" in {
+      val job = getUseCase(warnMaxExecutionTimeSeconds = Some(100))
+
+      val warnings = job.getTookTooLongWarnings(Instant.ofEpochSecond(1000), Instant.ofEpochSecond(1300), Some(200))
+
+      assert(warnings.nonEmpty)
+      assert(warnings.head == "The job took longer (00:05:00) than expected (00:01:40).")
+    }
+  }
+
   def getUseCase(tableDf: DataFrame = null,
                  dependencies: Seq[MetastoreDependency] = Nil,
                  isTableAvailable: Boolean = true,
                  isTableEmpty: Boolean = false,
-                 allowParallel: Boolean = true): JobBase = {
+                 allowParallel: Boolean = true,
+                 warnMaxExecutionTimeSeconds: Option[Int] = None): JobBase = {
     val operation = OperationDefFactory.getDummyOperationDef(dependencies = dependencies,
       allowParallel = allowParallel,
+      warnMaxExecutionTimeSeconds = warnMaxExecutionTimeSeconds,
       extraOptions = Map[String, String]("value" -> "7"))
 
     val bk = new SyncBookkeeperMock
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala
index e25fd3d1b..38f38e812 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/OperationDefSuite.scala
@@ -108,6 +108,7 @@ class OperationDefSuite extends AnyWordSpec with TempDirFixture {
            |class = "myclass"
            |output.table = "dummy_table"
            |always.attempt = "true"
+           |warn.maximum.execution.time.seconds = 50
            |
            |dependencies = [
            |  {
@@ -148,6 +149,7 @@ class OperationDefSuite extends AnyWordSpec with TempDirFixture {
       assert(op.dependencies.head.triggerUpdates)
       assert(op.dependencies(1).tables.contains("table2"))
       assert(!op.dependencies(1).triggerUpdates)
+      assert(op.warnMaxExecutionTimeSeconds.contains(50))
       assert(op.schemaTransformations.length == 4)
       assert(op.schemaTransformations.head.column == "A")
       assert(op.schemaTransformations.head.expression.contains("cast(A as decimal(15,5))"))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkTableSuite.scala
index fc636cb30..d1d77870c 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SinkTableSuite.scala
@@ -39,6 +39,7 @@ class SinkTableSuite extends AnyWordSpec {
           | },
           | {
           |  input.metastore.table = table44
+          |  warn.maximum.execution.time.seconds = 50
           |  transformations = [
           |    { col="a", expr="2+2" },
           |    { col="b", expr="cast(a) as string" },
@@ -80,8 +81,10 @@ class SinkTableSuite extends AnyWordSpec {
       assert(tbl3.outputTableName.contains("output_table"))
       assert(tbl3.rangeFromExpr.isEmpty)
       assert(tbl3.rangeToExpr.isEmpty)
+      assert(tbl3.warnMaxExecutionTimeSeconds.isEmpty)
 
       assert(tbl4.metaTableName == "table44")
+      assert(tbl4.warnMaxExecutionTimeSeconds.contains(50))
       assert(tbl4.transformations.length == 2)
       assert(tbl4.transformations.head.column == "a")
       assert(tbl4.transformations.head.expression.contains("2+2"))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SourceTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SourceTableSuite.scala
index 9c908ee4d..f1b24f2fa 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SourceTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/SourceTableSuite.scala
@@ -42,6 +42,7 @@ class SourceTableSuite extends AnyWordSpec {
           | {
           |  output.metastore.table = table44
           |  input.db.table = table14
+          |  warn.maximum.execution.time.seconds = 50
           |  transformations = [
           |    { col="a", expr="2+2" },
           |    { col="b", expr="cast(a) as string" },
@@ -85,10 +86,12 @@ class SourceTableSuite extends AnyWordSpec {
       assert(tbl3.filters.isEmpty)
       assert(tbl3.rangeFromExpr.isEmpty)
       assert(tbl3.rangeToExpr.isEmpty)
+      assert(tbl3.warnMaxExecutionTimeSeconds.isEmpty)
 
       assert(tbl4.metaTableName == "table44")
       assert(tbl4.query.isInstanceOf[Query.Table])
       assert(tbl4.query.asInstanceOf[Query.Table].dbTable == "table14")
+      assert(tbl4.warnMaxExecutionTimeSeconds.contains(50))
       assert(tbl4.transformations.length == 2)
       assert(tbl4.transformations.head.column == "a")
       assert(tbl4.transformations.head.expression.contains("2+2"))
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala
index a76640be0..a4be73520 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala
@@ -50,6 +50,7 @@ class TransferTableSuite extends AnyWordSpec {
           |      path = /datalake/path
           |      write.legacy = "true"
           |    }
+          |    warn.maximum.execution.time.seconds = 50
           |    transformations = [
           |      { col="a", expr="2+2" },
           |      { col="b", expr="cast(a) as string" },
@@ -115,6 +116,7 @@ class TransferTableSuite extends AnyWordSpec {
       assert(tbl3.filters.isEmpty)
       assert(tbl3.rangeFromExpr.isEmpty)
       assert(tbl3.rangeToExpr.isEmpty)
+      assert(tbl3.warnMaxExecutionTimeSeconds.isEmpty)
 
       assert(tbl4.query.isInstanceOf[Query.Table])
       assert(tbl4.query.asInstanceOf[Query.Table].dbTable == "db1.table2")
@@ -127,6 +129,7 @@ class TransferTableSuite extends AnyWordSpec {
       assert(tbl4.readOptions("mergeSchema") == "true")
       assert(tbl4.trackDays == 10)
       assert(tbl4.infoDateStart == LocalDate.of(2022, 1, 19))
+      assert(tbl4.warnMaxExecutionTimeSeconds.contains(50))
 
       assert(tbl4.transformations.length == 2)
       assert(tbl4.transformations.head.column == "a")