diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala index 79f603cfd..1f7b53165 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/jobrunner/ConcurrentJobRunnerImpl.scala +++ b/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 = { 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 cfd7f5ba0..dd1322085 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 @@ -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 } diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala index e9af6b018..b7f2042a0 100644 --- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ThreadUtils.scala +++ b/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) { diff --git a/pramen/core/src/test/resources/log4j2.properties b/pramen/core/src/test/resources/log4j2.properties index 315056667..c6ffee311 100644 --- a/pramen/core/src/test/resources/log4j2.properties +++ b/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 diff --git a/pramen/core/src/test/resources/test/config/integration_no_data_graceful.conf b/pramen/core/src/test/resources/test/config/integration_no_data_graceful.conf new file mode 100644 index 000000000..685848d93 --- /dev/null +++ b/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" + } + } +] diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/integration/NoDataGracefulSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/integration/NoDataGracefulSuite.scala new file mode 100644 index 000000000..9436550e3 --- /dev/null +++ b/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 + } + +} diff --git a/pramen/extras/src/test/resources/log4j2.properties b/pramen/extras/src/test/resources/log4j2.properties index fd3a3e791..12208729d 100644 --- a/pramen/extras/src/test/resources/log4j2.properties +++ b/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