#781 Add graceful handling of no data ingestions

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala

@@ -167,7 +167,15 @@ class ConcurrentJobRunnerImpl(runtimeConfig: RuntimeConfig,
       case _ => // skip
     }
 
-    statuses.forall(s => !s.isFailure)
+    statuses.forall { status =>
+      // This is to allow critical ingestion jobs stop the pipeline while not cause it to fail when
+      // `fail.if.no.data` is set to 'false'
+      val hasNoDataAsNotFailure = status match {
+        case RunStatus.NoData(failure) if !failure => job.operation.isCritical
+        case _ => false
+      }
+      !status.isFailure && !hasNoDataAsNotFailure
+    }
   }
 
   private[core] def runLazyJob(job: Job): Boolean = {

pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala

@@ -21,7 +21,7 @@ import com.fasterxml.jackson.module.scala.DefaultScalaModule
 import com.typesafe.config.Config
 import org.slf4j.{Logger, LoggerFactory}
 import sun.misc.Signal
-import za.co.absa.pramen.api.status.RunStatus.{NotRan, Succeeded}
+import za.co.absa.pramen.api.status.RunStatus.{NoData, NotRan, Succeeded}
 import za.co.absa.pramen.api.status._
 import za.co.absa.pramen.api.{NotificationBuilder, PipelineInfo, PipelineNotificationTarget, RunMode}
 import za.co.absa.pramen.core.app.config.RuntimeConfig.{DRY_RUN, EMAIL_IF_NO_CHANGES, UNDERCOVER}
@@ -489,7 +489,7 @@ object PipelineStateImpl {
       PipelineStatus.Success
     } else if (someTasksSucceeded && someTasksFailed && !strictFailures && !someCriticalTasksFailed) {
       PipelineStatus.PartialSuccess
-    } else if (someTasksSucceeded && !someTasksFailed && warningState) {
+    } else if (!someTasksFailed && warningState) {
       PipelineStatus.Warning
     } else {
       PipelineStatus.Failure
@@ -504,12 +504,15 @@ object PipelineStateImpl {
   }
 
   private def hasWarnings(taskResults: Seq[TaskResult], pipelineNotificationFailures: Seq[PipelineNotificationFailure]): Boolean = {
-    taskResults.exists{task =>
+    taskResults.exists { task =>
       val hasTaskWarnings = task.runStatus match {
         case succeeded: Succeeded =>
           val warnings = succeeded.warnings
             .filterNot(_.startsWith(SUPPRESS_WARNING_STARTING_WITH))
           warnings.nonEmpty
+        case noData: NoData =>
+          // The logic is this: if no data is not a failure, it is a warning.
+          !noData.isFailure
         case _ =>
           false
       }

pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala

@@ -79,14 +79,12 @@ object ThreadUtils {
       thread.join(waitMillis)
     }
 
+    val stackTrace = thread.getStackTrace
     val isAlive = thread.isAlive
-    val stackTrace = if (isAlive) {
-      val st = thread.getStackTrace
+    if (isAlive) {
       thread.interrupt()
       thread.join(closeWaitMillis)
-      st
-    } else
-      Array.empty[StackTraceElement]
+    }
 
     val closeableCount = ThreadClosableRegistry.getCloseableCount(threadId)
     if (closeableCount > 0) {

pramen/core/src/test/resources/log4j2.properties

@@ -11,12 +11,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-log4j.rootCategory=INFO, console
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.target=System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
-log4j.appender.console.Threshold=ERROR
+rootLogger.level=ERROR
+rootLogger.appenderRef.stdout.ref = console
+appender.console.type = Console
+appender.console.name = console
+appender.console.target = SYSTEM_ERR
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
 
 # Suppress warnings logged within various components
 logger.dummyprocessrunner.name = za.co.absa.pramen.core.mocks.DummyProcessRunner

pramen/core/src/test/resources/test/config/integration_no_data_graceful.conf

@@ -0,0 +1,118 @@
+# Copyright 2022 ABSA Group Limited
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This variable is expected to be set up by the test suite
+#base.path = "/tmp"
+
+pramen {
+  pipeline.name = "No data graceful test"
+
+  fail.if.no.data = false
+  bookkeeping.enabled = false
+  stop.spark.session = false
+}
+
+pramen.metastore {
+  tables = [
+    {
+      name = "table1"
+      format = "raw"
+      information.date.partition.by = overwrite
+      path = ${base.path}/table1
+    },
+    {
+      name = "table2"
+      format = "parquet"
+      information.date.partition.by = overwrite
+      path = ${base.path}/table2
+    },
+    {
+      name = "table3"
+      format = "parquet"
+      path = ${base.path}/table3
+    }
+  ]
+}
+
+pramen.sources.1 = [
+  {
+    name = "file_source"
+    factory.class = "za.co.absa.pramen.core.source.RawFileSource"
+  }
+]
+
+pramen.operations = [
+  {
+    name = "Sourcing from a folder"
+    type = "ingestion"
+    schedule.type = "daily"
+    critical = true
+
+    source = "file_source"
+
+    info.date.expr = "@runDate"
+
+    tables = [
+      {
+        input.path = ${base.path}
+        output.metastore.table = table1
+      }
+    ]
+  },
+  {
+    name = "Converting to parquet"
+    type = "transformation"
+
+    class = "za.co.absa.pramen.core.transformers.ConversionTransformer"
+    schedule.type = "daily"
+
+    output.table = "table2"
+
+    dependencies = [
+      {
+        tables = [ table1 ]
+        date.from = "@infoDate"
+        optional = true # Since no bookkeeping available the table will be seen as empty for the dependency manager
+      }
+    ]
+
+    option {
+      input.table = "table1"
+      input.format = "csv"
+      use.file.list = ${use.file.list}
+
+      header = true
+    }
+  },
+  {
+    name = "Anothe rtransformation"
+    type = "transformation"
+
+    class = "za.co.absa.pramen.core.transformers.IdentityTransformer"
+    schedule.type = "daily"
+
+    output.table = "table3"
+
+    dependencies = [
+      {
+        tables = [ table2 ]
+        date.from = "@infoDate"
+        optional = true # Since no bookkeeping available the table will be seen as empty for the dependency manager
+      }
+    ]
+
+    option {
+      input.table = "table2"
+    }
+  }
+]

pramen/core/src/test/scala/za/co/absa/pramen/core/integration/NoDataGracefulSuite.scala

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.core.integration
+
+import com.typesafe.config.{Config, ConfigFactory}
+import org.apache.hadoop.fs.Path
+import org.scalatest.wordspec.AnyWordSpec
+import org.slf4j.LoggerFactory
+import za.co.absa.pramen.core.base.SparkTestBase
+import za.co.absa.pramen.core.fixtures.{TempDirFixture, TextComparisonFixture}
+import za.co.absa.pramen.core.runner.AppRunner
+import za.co.absa.pramen.core.utils.{FsUtils, ResourceUtils}
+
+import java.time.LocalDate
+
+class NoDataGracefulSuite extends AnyWordSpec with SparkTestBase with TempDirFixture with TextComparisonFixture {
+  private val log = LoggerFactory.getLogger(this.getClass)
+
+  private val infoDate = LocalDate.of(2021, 2, 18)
+
+  "Graceful no data handling" should {
+    "work end to end when data is available" in {
+      withTempDirectory("integration_file_based") { tempDir =>
+        val fsUtils = new FsUtils(spark.sparkContext.hadoopConfiguration, tempDir)
+
+        fsUtils.writeFile(new Path(tempDir, "landing_file1.csv"), "id,name\n1,John\n2,Jack\n3,Jill\n")
+        fsUtils.writeFile(new Path(tempDir, "landing_file2.csv"), "id,name\n4,Mary\n5,Jane\n6,Kate\n")
+
+        log.info("test")
+        val conf = getConfig(tempDir)
+
+        val exitCode = AppRunner.runPipeline(conf)
+
+        assert(exitCode == 0)
+
+        val table2Path = new Path(tempDir, "table2")
+        val df = spark.read.parquet(table2Path.toString)
+
+        assert(!df.isEmpty)
+      }
+    }
+
+    "succeeds with warning when data is not available" in {
+      withTempDirectory("integration_file_based") { tempDir =>
+        val conf = getConfig(tempDir)
+
+        val exitCode = AppRunner.runPipeline(conf)
+
+        assert(exitCode == 0)
+      }
+    }
+  }
+
+  def getConfig(basePath: String, useFileList: Boolean = false): Config = {
+    val configContents = ResourceUtils.getResourceString("/test/config/integration_no_data_graceful.conf")
+    val basePathEscaped = basePath.replace("\\", "\\\\")
+
+    val conf = ConfigFactory.parseString(
+      s"""base.path = "$basePathEscaped"
+         |use.file.list = $useFileList
+         |pramen.runtime.is.rerun = true
+         |pramen.current.date = "$infoDate"
+         |$configContents
+         |""".stripMargin
+    ).withFallback(ConfigFactory.load())
+      .resolve()
+
+    conf
+  }
+
+}

pramen/extras/src/test/resources/log4j2.properties

@@ -12,12 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-log4j.rootCategory=INFO, console
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.target=System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
-log4j.appender.console.Threshold=ERROR
+rootLogger.level = ERROR
+rootLogger.appenderRef.stdout.ref = console
+appender.console.type = Console
+appender.console.name = console
+appender.console.target = SYSTEM_ERR
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
 
 # Suppress warnings logged within various components
 logger.dummyprocessrunner.name = za.co.absa.pramen.core.mocks.DummyProcessRunner