Skip to content

#685 Fix when transformers and sinks are marked as not allowed to run in parallel have the behavior of transformations with self-dependencies. #688

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yruslan merged 1 commit into from
Jan 19, 2026
Merged

#685 Fix when transformers and sinks are marked as not allowed to run in parallel have the behavior of transformations with self-dependencies. #688

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Job.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ trait Job {

def allowRunningTasksInParallel: Boolean

def isSelfDependent: Boolean
Copy link

@coderabbitai coderabbitai bot Jan 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Avoid breaking external Job implementations with new abstract API.

Adding a new abstract member to a public trait is binary/source breaking for downstream implementers that aren’t updated/recompiled. Consider providing a safe default (e.g., false) or explicitly documenting the required migration in release notes. Line 45.

💡 Suggested compatibility-friendly default 
-  def isSelfDependent: Boolean
+  def isSelfDependent: Boolean = false

To verify in-repo implementers:

#!/bin/bash
# Find Job implementations that may need the new member.
rg -nP --type=scala '\bextends\s+Job\b|\bwith\s+Job\b'
🤖 Prompt for AI Agents 
In `@pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/Job.scala` at line
45, The new abstract member isSelfDependent on trait Job breaks downstream
implementations; provide a compatibility-safe default by making isSelfDependent
a concrete method returning false in the Job trait (i.e., change the declaration
in Job so implementations keep working while allowing overrides), and update any
internal implementers that need true to explicitly override isSelfDependent.


def notificationTargets: Seq[JobNotificationTarget]

def trackDays: Int
Expand Down
2 changes: 2 additions & 0 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/JobBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ abstract class JobBase(operationDef: OperationDef,

override val allowRunningTasksInParallel: Boolean = operationDef.allowParallel && !hasSelfDependencies

override val isSelfDependent: Boolean = hasSelfDependencies

override def notificationTargets: Seq[JobNotificationTarget] = jobNotificationTargets

override def trackDays: Int = outputTable.trackDays
Expand Down
9 changes: 5 additions & 4 deletions pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,17 +107,18 @@ abstract class TaskRunnerBase(conf: Config,
val sortedTasks = tasks.sortBy(_.infoDate)
var failedInfoDate: Option[LocalDate] = None

sortedTasks.map(task =>
sortedTasks.map { task =>
val selfDependent = task.job.isSelfDependent
failedInfoDate match {
case Some(failedDate) =>
case Some(failedDate) if selfDependent =>
skipTask(task, s"Due to failure for $failedDate", isWarning = true)
case None =>
case _ =>
val status = runTask(task)
if (status.isFailure)
failedInfoDate = Option(task.infoDate)
status
}
)
}
}

/** Runs a task in the single thread. Performs all task logging and notification sending activities. */
Expand Down
5 changes: 4 additions & 1 deletion pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/job/JobSpy.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ class JobSpy(jobName: String = "Dummy Job",
runFunction: () => RunResult = () => null,
scheduleStrategyIn: ScheduleStrategy = new ScheduleStrategySourcing(true),
allowParallel: Boolean = true,
hasSelfDependencies: Boolean = false,
saveStats: MetaTableStats = MetaTableStats(Some(0)),
jobNotificationTargets: Seq[JobNotificationTarget] = Seq.empty,
jobTrackDays: Int = 0
Expand All @@ -66,7 +67,9 @@ class JobSpy(jobName: String = "Dummy Job",

override val scheduleStrategy: ScheduleStrategy = scheduleStrategyIn

override def allowRunningTasksInParallel: Boolean = allowParallel
override def allowRunningTasksInParallel: Boolean = allowParallel && !hasSelfDependencies

override def isSelfDependent: Boolean = hasSelfDependencies

override def notificationTargets: Seq[JobNotificationTarget] = jobNotificationTargets

Expand Down
35 changes: 35 additions & 0 deletions ...en/core/src/test/scala/za/co/absa/pramen/core/tests/runner/task/TaskRunnerBaseSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,39 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar

val job = tasks.head.job.asInstanceOf[JobSpy]

assert(job.validateCount == 2)
assert(job.runCount == 2)
assert(job.postProcessingCount == 0)
assert(job.saveCount == 0)
assert(job.createHiveTableCount == 0)
assert(result.length == 2)
assert(result.head.runStatus.isInstanceOf[Failed])
assert(result(1).runStatus.isInstanceOf[Failed])

val journalEntries = journal.getEntries(now, now.plusSeconds(30))

assert(journalEntries.length == 2)
assert(journalEntries.head.status == "Failed")
assert(notificationTarget.notificationsSent.length == 2)
assert(notificationTarget.notificationsSent.head.runStatus.isInstanceOf[RunStatus.Failed])
}

"run multiple failure jobs sequential execution and self=dependencies" in {
val now = Instant.now()
val notificationTarget = new NotificationTargetSpy(ConfigFactory.empty(), (action: TaskResult) => ())
val jobNotificationTarget = JobNotificationTarget("notification1", Map.empty[String, String], notificationTarget)
val (runner, _, journal, state, _, tasks) = getUseCase(hasSelfDependencies = true, runFunction = () => throw new IllegalStateException("Test exception"), jobNotificationTargets = Seq(jobNotificationTarget))

val taskPreDefs = (infoDate :: infoDate.plusDays(1) :: Nil).map(d => core.pipeline.TaskPreDef(d, 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)
Expand Down Expand Up @@ -648,6 +681,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
isRerun: Boolean = false,
bookkeeperIn: Bookkeeper = null,
allowParallel: Boolean = true,
hasSelfDependencies: Boolean = false,
hiveTable: Option[String] = None,
jobNotificationTargets: Seq[JobNotificationTarget] = Nil,
timeoutTask: Boolean = false
Expand Down Expand Up @@ -675,6 +709,7 @@ class TaskRunnerBaseSuite extends AnyWordSpec with SparkTestBase with TextCompar
runFunction = runFunction,
operationDef = operationDef,
allowParallel = allowParallel,
hasSelfDependencies = hasSelfDependencies,
saveStats = stats,
hiveTable = hiveTable,
jobNotificationTargets = jobNotificationTargets)
Expand Down
Loading