Repository: AbsaOSS/pramen (language: python)

Issue:
Add spark_application_id to journal
## Background
This would help generating reports based on `journal` table.

## Feature
Add spark_application_id to journal.

## Example [Optional]
--

## Proposed Solution [Optional]
Also add:
- Pipeline ID (generate on fly)
- Pipeline name
- Environment
- Tenant



Discussion / hints:
I'll take this one and also prepare the concept of bookkeeping. 
What we want to achieve looks very similar to the proper bookkeeping pattern being used for monitoring and compliance purpose.  
Sure!

I didn't add much context to this issue, please reach me out if you have questions.

Fix patch:
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala
index 12fae8755..83fa80a59 100644
--- a/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala
+++ b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala
@@ -28,5 +28,7 @@ case class PipelineInfo(
                          finishedAt: Option[Instant],
                          sparkApplicationId: Option[String],
                          failureException: Option[Throwable],
-                         pipelineNotificationFailures: Seq[PipelineNotificationFailure]
+                         pipelineNotificationFailures: Seq[PipelineNotificationFailure],
+                         pipelineId: String,
+                         tenant: Option[String]
                        )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala
index 849a1ae6e..45488735d 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalHadoop.scala
@@ -81,7 +81,12 @@ class JournalHadoop(journalPath: String)
         startedAt = v.startedAt,
         finishedAt = v.finishedAt,
         status = v.status,
-        failureReason = v.failureReason
+        failureReason = v.failureReason,
+        sparkApplicationId = v.sparkApplicationId,
+        pipelineId = v.pipelineId,
+        pipelineName = v.pipelineName,
+        environmentName = v.environmentName,
+        tenant = v.tenant
       ))
   }
 
@@ -108,6 +113,11 @@ class JournalHadoop(journalPath: String)
       t.finishedAt ::
       t.status ::
       removeSeparators(t.failureReason.getOrElse("")) ::
+      t.sparkApplicationId.getOrElse("") ::
+      t.pipelineId.getOrElse("") ::
+      t.pipelineName.getOrElse("") ::
+      t.environmentName.getOrElse("") ::
+      t.tenant.getOrElse("") ::
       Nil
     record.mkString("", s"$separator", "\n")
   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalJdbc.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalJdbc.scala
index a94332abc..7346581be 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalJdbc.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/JournalJdbc.scala
@@ -49,7 +49,12 @@ class JournalJdbc(db: Database) extends Journal {
       entry.startedAt,
       entry.finishedAt,
       entry.status,
-      entry.failureReason)
+      entry.failureReason,
+      entry.sparkApplicationId,
+      entry.pipelineId,
+      entry.pipelineName,
+      entry.environmentName,
+      entry.tenant)
 
     try {
       db.run(
@@ -81,7 +86,12 @@ class JournalJdbc(db: Database) extends Journal {
         startedAt = v.startedAt,
         finishedAt = v.finishedAt,
         status = v.status,
-        failureReason = v.failureReason
+        failureReason = v.failureReason,
+        sparkApplicationId = v.sparkApplicationId,
+        pipelineId = v.pipelineId,
+        pipelineName = v.pipelineName,
+        environmentName = v.environmentName,
+        tenant = v.tenant
       )
     }).toList
   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTask.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTask.scala
index 2ebfa035b..a0bb1d22d 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTask.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTask.scala
@@ -30,5 +30,10 @@ case class JournalTask(
                           startedAt: Long,
                           finishedAt: Long,
                           status: String,
-                          failureReason: Option[String]
+                          failureReason: Option[String],
+                          sparkApplicationId: Option[String],
+                          pipelineId: Option[String],
+                          pipelineName: Option[String],
+                          environmentName: Option[String],
+                          tenant: Option[String]
                         )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTasks.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTasks.scala
index e06337221..0e76da77c 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTasks.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/JournalTasks.scala
@@ -34,9 +34,14 @@ class JournalTasks(tag: Tag) extends Table[JournalTask](tag, "journal") {
   def finishedAt = column[Long]("finished_at")
   def status = column[String]("status", O.Length(50))
   def failureReason = column[Option[String]]("failure_reason")
+  def sparkApplicationId = column[Option[String]]("spark_application_id", O.Length(128))
+  def pipelineId = column[Option[String]]("pipelineId", O.Length(40))
+  def pipelineName = column[Option[String]]("pipelineName", O.Length(200))
+  def environmentName = column[Option[String]]("environmentName", O.Length(128))
+  def tenant = column[Option[String]]("tenant", O.Length(200))
   def * = (jobName, pramenTableName, periodBegin, periodEnd,
     informationDate, inputRecordCount, inputRecordCountOld, outputRecordCount,
-    outputRecordCountOld, outputSize, startedAt, finishedAt, status, failureReason) <> (JournalTask.tupled, JournalTask.unapply)
+    outputRecordCountOld, outputSize, startedAt, finishedAt, status, failureReason, sparkApplicationId, pipelineId, pipelineName, environmentName, tenant) <> (JournalTask.tupled, JournalTask.unapply)
   def idx1 = index("idx_started_at", startedAt, unique = false)
   def idx2 = index("idx_finished_at", finishedAt, unique = false)
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala
index e857f71a8..a39fec579 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompleted.scala
@@ -16,6 +16,7 @@
 
 package za.co.absa.pramen.core.journal.model
 
+import za.co.absa.pramen.api.PipelineInfo
 import za.co.absa.pramen.api.status.RunStatus.Succeeded
 import za.co.absa.pramen.api.status.TaskResult
 import za.co.absa.pramen.core.pipeline.Task
@@ -36,7 +37,12 @@ case class TaskCompleted(
                           startedAt: Long,
                           finishedAt: Long,
                           status: String,
-                          failureReason: Option[String]
+                          failureReason: Option[String],
+                          sparkApplicationId: Option[String],
+                          pipelineId: Option[String],
+                          pipelineName: Option[String],
+                          environmentName: Option[String],
+                          tenant: Option[String]
                         )
 
 object TaskCompleted {
@@ -44,12 +50,13 @@ object TaskCompleted {
     * TaskCompleted is a legacy form of a task completion status. This is added for compatibility.
     * The journal might be eventually deprecated.
     */
-  def fromTaskResult(task: Task, taskResult: TaskResult): TaskCompleted = {
+  def fromTaskResult(task: Task, taskResult: TaskResult, pipelineInfo: PipelineInfo): TaskCompleted = {
     val now = Instant.now().getEpochSecond
     val taskStarted = taskResult.runInfo.map(_.started.getEpochSecond).getOrElse(now)
     val taskFinished = taskResult.runInfo.map(_.finished.getEpochSecond).getOrElse(now)
     val status = taskResult.runStatus.toString
     val failureReason = taskResult.runStatus.getReason
+    val sparkApplicationId = Option(taskResult.applicationId)
 
     val (recordCountOld, inputRecordCount, outputRecordCount, sizeBytes) = taskResult.runStatus match {
       case s: Succeeded => (s.recordCountOld, s.recordCount, Some(s.recordCount), s.sizeBytes)
@@ -70,7 +77,12 @@ object TaskCompleted {
       taskStarted,
       taskFinished,
       status,
-      failureReason
+      failureReason,
+      sparkApplicationId,
+      Option(pipelineInfo.pipelineId),
+      Option(pipelineInfo.pipelineName),
+      Option(pipelineInfo.environment),
+      pipelineInfo.tenant
     )
   }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompletedCsv.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompletedCsv.scala
index 8b6dd561e..3aa1bd128 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompletedCsv.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/journal/model/TaskCompletedCsv.scala
@@ -30,5 +30,10 @@ case class TaskCompletedCsv(
                              startedAt: Long,
                              finishedAt: Long,
                              status: String,
-                             failureReason: Option[String]
+                             failureReason: Option[String],
+                             sparkApplicationId: Option[String],
+                             pipelineId: Option[String],
+                             pipelineName: Option[String],
+                             environmentName: Option[String],
+                             tenant: Option[String]
                            )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PipelineDef.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PipelineDef.scala
index fe25f6474..90687ce64 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PipelineDef.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/PipelineDef.scala
@@ -24,13 +24,15 @@ import za.co.absa.pramen.core.utils.ConfigUtils
 case class PipelineDef(
                         name: String,
                         environment: String,
-                        operations: Seq[OperationDef]
+                        operations: Seq[OperationDef],
+                        tenant: Option[String]
                       )
 
 object PipelineDef {
   val ENVIRONMENT_NAME = "pramen.environment.name"
   val PIPELINE_NAME_KEY = "pramen.pipeline.name"
   val OPERATIONS_KEY = "pramen.operations"
+  val TENANT_KEY = "pramen.tenant"
 
   def fromConfig(conf: Config, infoDateConfig: InfoDateConfig): PipelineDef = {
     val defaultDelayDays = conf.getInt(EXPECTED_DELAY_DAYS)
@@ -40,6 +42,7 @@ object PipelineDef {
       .zipWithIndex
       .flatMap{ case (c, i) => OperationDef.fromConfig(c, conf, infoDateConfig, s"$OPERATIONS_KEY[$i]", defaultDelayDays) }
       .toSeq
-    PipelineDef(name, environment, operations)
+    val tenant = ConfigUtils.getOptionString(conf, TENANT_KEY)
+    PipelineDef(name, environment, operations, tenant)
   }
 }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/rdb/PramenDb.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/rdb/PramenDb.scala
index b2ae7b3ee..914a2a997 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/rdb/PramenDb.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/rdb/PramenDb.scala
@@ -63,6 +63,14 @@ class PramenDb(val jdbcConfig: JdbcConfig,
     if (dbVersion < 3) {
       initTable(MetadataRecords.records.schema)
     }
+
+    if (0 < dbVersion && dbVersion < 4) {
+      addColumn(JournalTasks.journalTasks.baseTableRow.tableName, "spark_application_id", "varchar(128)")
+      addColumn(JournalTasks.journalTasks.baseTableRow.tableName, "pipelineId", "varchar(40)")
+      addColumn(JournalTasks.journalTasks.baseTableRow.tableName, "pipelineName", "varchar(200)")
+      addColumn(JournalTasks.journalTasks.baseTableRow.tableName, "environmentName", "varchar(128)")
+      addColumn(JournalTasks.journalTasks.baseTableRow.tableName, "tenant", "varchar(200)")
+    }
   }
 
   def initTable(schema: H2Profile.SchemaDescription): Unit = {
@@ -77,6 +85,20 @@ class PramenDb(val jdbcConfig: JdbcConfig,
     }
   }
 
+  def addColumn(table: String, columnName: String, columnType: String): Unit = {
+    try {
+      val quotedTable = s""""$table""""
+      val quotedColumnName = s""""$columnName""""
+      db.run(
+          sqlu"ALTER TABLE #$quotedTable ADD #$quotedColumnName #$columnType"
+        ).execute()
+    } catch {
+      case NonFatal(ex) =>
+        throw new RuntimeException(s"Unable to add column: '${columnName} ${columnType}' to table: '${table}'for the url: $activeUrl", ex)
+    }
+  }
+
+
   override def close(): Unit = {
     jdbcConnection.close()
     slickDb.close()
@@ -84,7 +106,7 @@ class PramenDb(val jdbcConfig: JdbcConfig,
 }
 
 object PramenDb {
-  val MODEL_VERSION = 3
+  val MODEL_VERSION = 4
   val DEFAULT_RETRIES = 3
 
   def apply(jdbcConfig: JdbcConfig): PramenDb = {
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 7d69cc378..565ff4c51 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
@@ -446,13 +446,13 @@ abstract class TaskRunnerBase(conf: Config,
 
     logTaskResult(updatedResult, isLazy)
     pipelineState.addTaskCompletion(Seq(updatedResult))
-    addJournalEntry(task, updatedResult)
+    addJournalEntry(task, updatedResult, pipelineState.getState().pipelineInfo)
 
     updatedResult.runStatus
   }
 
-  private def addJournalEntry(task: Task, taskResult: TaskResult): Unit = {
-    val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult)
+  private def addJournalEntry(task: Task, taskResult: TaskResult, pipelineInfo: PipelineInfo): Unit = {
+    val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult, pipelineInfo)
 
     journal.addEntry(taskCompleted)
   }
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 b5dcf6b7a..fccc62af7 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
@@ -42,8 +42,10 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
   protected val log: Logger = LoggerFactory.getLogger(this.getClass)
 
   // Config
+  private val pipelineId = java.util.UUID.randomUUID.toString
   private val pipelineName = conf.getString(PIPELINE_NAME_KEY)
   private val environmentName = conf.getString(ENVIRONMENT_NAME)
+  private val tenant = ConfigUtils.getOptionString(conf, TENANT_KEY)
   private val sendEmailIfNoNewData: Boolean = conf.getBoolean(EMAIL_IF_NO_CHANGES)
   private val hookConfig = HookConfig.fromConfig(conf)
   private var pipelineNotificationTargets: Seq[PipelineNotificationTarget] = Seq.empty
@@ -108,7 +110,9 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
         finishedInstant,
         sparkAppId,
         appException,
-        pipelineNotificationFailures.toSeq
+        pipelineNotificationFailures.toSeq,
+        pipelineId,
+        tenant
       ),
       isFinished,
       warningFlag,


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala
index 894c2273d..c5b06e88d 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala
@@ -29,7 +29,8 @@ object PipelineInfoFactory {
                            finishedAt: Option[Instant] = None,
                            sparkApplicationId: Option[String] = Some("testid-12345"),
                            failureException: Option[Throwable] = None,
-                           pipelineNotificationFailures: Seq[PipelineNotificationFailure] = Seq.empty): PipelineInfo = {
-    PipelineInfo(pipelineName, environment, runtimeInfo, startedAt, finishedAt, sparkApplicationId, failureException, pipelineNotificationFailures)
+                           pipelineNotificationFailures: Seq[PipelineNotificationFailure] = Seq.empty,
+                           tenant:  Option[String] = Some("Dummy tenant")): PipelineInfo = {
+    PipelineInfo(pipelineName, environment, runtimeInfo, startedAt, finishedAt, sparkApplicationId, failureException, pipelineNotificationFailures, java.util.UUID.randomUUID().toString,tenant)
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala
index 6ca935fbd..7c9f43cf9 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TaskCompletedFactory.scala
@@ -23,7 +23,7 @@ import za.co.absa.pramen.core.journal.model.TaskCompleted
 import java.time.LocalDate
 
 object TaskCompletedFactory {
-  def getTackCompleted(jobName: String = "DummyJob",
+  def getTaskCompleted(jobName: String = "DummyJob",
                        tableName: String = "DummyTable",
                        periodBegin: LocalDate = LocalDate.of(2020, 12, 9),
                        periodEnd: LocalDate = LocalDate.of(2020, 12, 9),
@@ -36,7 +36,12 @@ object TaskCompletedFactory {
                        startedAt: Long = 1234567L,
                        finishedAt: Long = 1234568L,
                        status: String = TaskStatus.NEW.toString,
-                       failureReason: Option[String] = None): TaskCompleted = {
+                       failureReason: Option[String] = None,
+                       sparkApplicationId: Option[String] = Some("abc123"),
+                       pipelineId: Option[String] = Some(java.util.UUID.randomUUID().toString),
+                       pipelineName: Option[String] = Some("test"),
+                       environmentName: Option[String] = Some("DEV"),
+                       tenant: Option[String] = Some("Dummy tenant")): TaskCompleted = {
     model.TaskCompleted(
       jobName,
       tableName,
@@ -51,7 +56,12 @@ object TaskCompletedFactory {
       startedAt,
       finishedAt,
       status,
-      failureReason
+      failureReason,
+      sparkApplicationId,
+      pipelineId,
+      pipelineName,
+      environmentName,
+      tenant
     )
   }
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala
index d71193310..d799e2f42 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TaskCompletedSuite.scala
@@ -16,8 +16,10 @@
 
 package za.co.absa.pramen.core.tests.journal
 
+import org.apache.avro.generic.GenericData.Array
 import org.scalatest.wordspec.AnyWordSpec
-import za.co.absa.pramen.api.status.{RunInfo, RunStatus, TaskResult, TaskRunReason}
+import za.co.absa.pramen.api.PipelineInfo
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, RunInfo, RunStatus, RuntimeInfo, TaskResult, TaskRunReason}
 import za.co.absa.pramen.core.journal.model.TaskCompleted
 import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.mocks.job.JobSpy
@@ -27,6 +29,17 @@ import java.time.{Instant, LocalDate}
 
 class TaskCompletedSuite extends AnyWordSpec {
   private val infoDate = LocalDate.of(2022, 1, 18)
+  private val pipelineInfo = PipelineInfo(
+    pipelineName = "test_pipeline",
+    environment = "TEST",
+    runtimeInfo = RuntimeInfo(),
+    startedAt = Instant.now(),
+    finishedAt = Some(Instant.now()),
+    sparkApplicationId = Some("app_123"),
+    failureException = None,
+    pipelineNotificationFailures = Seq.empty,
+    pipelineId = "test_pipeline_id",
+    tenant = Some("test_tenant"))
 
   "fromTaskResult" should {
     "create a TaskCompleted from a TaskResult for a successful task" in {
@@ -46,8 +59,7 @@ class TaskCompletedSuite extends AnyWordSpec {
         Nil,
         Nil,
         Map.empty)
-
-      val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult)
+      val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult, pipelineInfo)
 
       assert(taskCompleted.jobName == job.name)
       assert(taskCompleted.tableName == job.outputTable.name)
@@ -62,7 +74,11 @@ class TaskCompletedSuite extends AnyWordSpec {
       assert(taskCompleted.startedAt == now.minusSeconds(10).getEpochSecond)
       assert(taskCompleted.finishedAt == now.getEpochSecond)
       assert(taskCompleted.status == "Update")
-      assert(taskCompleted.failureReason.isEmpty)
+      assert(taskCompleted.sparkApplicationId == pipelineInfo.sparkApplicationId)
+      assert(taskCompleted.pipelineId.contains(pipelineInfo.pipelineId))
+      assert(taskCompleted.pipelineName.contains(pipelineInfo.pipelineName))
+      assert(taskCompleted.environmentName.contains(pipelineInfo.environment))
+      assert(taskCompleted.tenant == pipelineInfo.tenant)
     }
 
     "create a TaskCompleted from a TaskResult for a failed task" in {
@@ -83,7 +99,7 @@ class TaskCompletedSuite extends AnyWordSpec {
         Nil,
         Map.empty)
 
-      val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult)
+      val taskCompleted = TaskCompleted.fromTaskResult(task, taskResult, pipelineInfo)
 
       assert(taskCompleted.jobName == job.name)
       assert(taskCompleted.tableName == job.outputTable.name)
@@ -99,6 +115,11 @@ class TaskCompletedSuite extends AnyWordSpec {
       assert(now.getEpochSecond - taskCompleted.finishedAt < 1000)
       assert(taskCompleted.status == "Failed")
       assert(taskCompleted.failureReason.exists(_.contains("Dummy Exception")))
+      assert(taskCompleted.sparkApplicationId == pipelineInfo.sparkApplicationId)
+      assert(taskCompleted.pipelineId.contains(pipelineInfo.pipelineId))
+      assert(taskCompleted.pipelineName.contains(pipelineInfo.pipelineName))
+      assert(taskCompleted.environmentName.contains(pipelineInfo.environment))
+      assert(taskCompleted.tenant == pipelineInfo.tenant)
     }
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TestCases.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TestCases.scala
index a9f70df1a..a2672320d 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TestCases.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/journal/TestCases.scala
@@ -30,7 +30,7 @@ object TestCases {
   val instant2: Instant = Instant.ofEpochSecond(1597318835)
   val instant3: Instant = Instant.ofEpochSecond(1597318839)
 
-  val task1: TaskCompleted = model.TaskCompleted("job1", "table1", infoDate1, infoDate1, infoDate1, 100, 0, Some(100), None, None, 597318830, 1597318830, "New", Some("Test1"))
-  val task2: TaskCompleted = model.TaskCompleted("job1", "table1", infoDate2, infoDate2, infoDate2, 100, 0, Some(100), None, None, 1597318835, 1597318835, "Late", Some("Test2"))
-  val task3: TaskCompleted = model.TaskCompleted("job2", "table2", infoDate3, infoDate3, infoDate3, 100, 0, Some(100), None, None, 1597318839, 1597318839, "Fail", Some("Test3"))
+  val task1: TaskCompleted = model.TaskCompleted("job1", "table1", infoDate1, infoDate1, infoDate1, 100, 0, Some(100), None, None, 597318830, 1597318830, "New", Some("Test1"), Some("abc123"), Some("p_id_1"), Some("p_1"), Some("TEST"), Some("T1"))
+  val task2: TaskCompleted = model.TaskCompleted("job1", "table1", infoDate2, infoDate2, infoDate2, 100, 0, Some(100), None, None, 1597318835, 1597318835, "Late", Some("Test2"), Some("abc123"), Some("p_id_2"), Some("p_2"), Some("TEST"), Some("T2"))
+  val task3: TaskCompleted = model.TaskCompleted("job2", "table2", infoDate3, infoDate3, infoDate3, 100, 0, Some(100), None, None, 1597318839, 1597318839, "Fail", Some("Test3"), Some("abc123"), Some("p_id_3"), Some("p_2"), Some("TEST"), Some("T2"))
 }
diff --git a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
index 0320bc883..0e71549b4 100644
--- a/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
+++ b/pramen/extras/src/test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
@@ -55,7 +55,7 @@ class EcsPipelineNotificationTargetSuite extends AnyWordSpec {
       val task3 = TestPrototypes.taskNotification.copy(jobName = "Job 3", outputTable = metaTableDef3)
 
       notificationTarget.sendNotification(
-        PipelineInfo("Dummy", "DEV", RuntimeInfo(), Instant.now, None, None, None, Seq.empty),
+        PipelineInfo("Dummy", "DEV", RuntimeInfo(), Instant.now, None, None, None, Seq.empty, "pid_123", None),
         Seq(task1, task2, task3),
         CustomNotification(Seq.empty, Seq.empty)
       )