From e00c7d6888b408b4f2afb47a1db6c81c2b8b674a Mon Sep 17 00:00:00 2001 From: Dogukan Cavdaroglu Date: Wed, 30 Sep 2026 17:43:01 +0300 Subject: [PATCH] :sparkles: feat(engine): add skipWrite execution option to map without writing to the sink --- README.md | 1 + .../main/scala/io/ignifyr/engine/Boot.scala | 7 ++- .../engine/cli/CommandLineInterface.scala | 14 +++++- .../engine/data/write/SinkHandler.scala | 10 ++++- .../engine/execution/MappingJobLauncher.scala | 10 +++++ .../execution/log/ExecutionLogger.scala | 2 + .../processing/FileStreamInputArchiver.scala | 7 ++- .../mapping/job/FhirMappingJobManager.scala | 6 ++- .../model/FhirMappingJobExecution.scala | 13 ++++-- .../engine/model/FhirMappingJobResult.scala | 4 +- .../engine/cli/CommandLineInterfaceTest.scala | 14 ++++++ .../test/engine/data/SinkHandlerTest.scala | 26 ++++++++++- .../FileStreamInputArchiverTest.scala | 18 ++++++++ .../MappingJobLauncherSkipWriteTest.scala | 45 +++++++++++++++++++ .../model/FhirMappingJobExecutionTest.scala | 27 ++++++++++- 15 files changed, 189 insertions(+), 15 deletions(-) create mode 100644 ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/MappingJobLauncherSkipWriteTest.scala diff --git a/README.md b/README.md index 6e8d17c5..de007086 100644 --- a/README.md +++ b/README.md @@ -194,6 +194,7 @@ When started as a standalone tool, the engine can run in two modes based on argu - `run`: Runs a configured mapping-job as a batch process and shuts down. `run` command accepts the following parameters: - `--job`: The path to the mapping-job to be executed. If provided, overrides the path provided to the JVM as the configuration parameter. - `--db` (alias `--db-path`): The path to the database folder that is used for scheduled jobs. If provided, overrides the path provided to the JVM as the configuration parameter. + - `--skip-write` (no value): Runs the mappings and logs their results, but writes nothing to the sink, does not validate the sink and does not archive the source files. Use it to test a job. Only batch jobs support it; a streaming or scheduled job is rejected. The mappings and schemas folders are not command-line options; they are always read from the [configuration](#configurations) (`ignifyr.mappings.repository.folder-path` and diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/Boot.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/Boot.scala index ad1d9b75..2ba430ab 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/Boot.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/Boot.scala @@ -38,7 +38,12 @@ object Boot extends App { if (options.contains("db-path")) options("db-path").asInstanceOf[String] else IgnifyrConfig.engineConfig.ignifyrDbFolderPath - CommandLineInterface.runJob(ignifyrEngine, mappingJobFilePath, ignifyrDbFolderPath) + CommandLineInterface.runJob( + ignifyrEngine, + mappingJobFilePath, + ignifyrDbFolderPath, + skipWrite = options.contains("skip-write") + ) // Diagnostic: list the installed extension modules and everything they contribute. Reads only // the ServiceLoader-backed registry, so it deliberately does NOT build the engine (no workspace diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/cli/CommandLineInterface.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/cli/CommandLineInterface.scala index 79fd1a64..ad341930 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/cli/CommandLineInterface.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/cli/CommandLineInterface.scala @@ -84,14 +84,21 @@ object CommandLineInterface { * * @param ignifyrEngine * @param mappingJobFilePath + * @param skipWrite Run the mappings without writing their results to the sink (batch jobs only) */ - def runJob(ignifyrEngine: IgnifyrEngine, mappingJobFilePath: Option[String], ignifyrDbFolderPath: String): Unit = { + def runJob( + ignifyrEngine: IgnifyrEngine, + mappingJobFilePath: Option[String], + ignifyrDbFolderPath: String, + skipWrite: Boolean = false + ): Unit = { if (mappingJobFilePath.isEmpty) { println("There are no jobs to run. Exiting...") System.exit(1) } val mappingJob = FhirMappingJobFormatter.readMappingJobFromFile(mappingJobFilePath.get) - val mappingJobExecution = FhirMappingJobExecution(job = mappingJob, mappingTasks = mappingJob.mappings) + val mappingJobExecution = + FhirMappingJobExecution(job = mappingJob, mappingTasks = mappingJob.mappings, skipWrite = skipWrite) new MappingJobLauncher(ignifyrEngine).launch(mappingJob, mappingJobExecution, ignifyrDbFolderPath) match { case MappingJobLaunch.Batch(completion) => Await.result(completion, Duration.Inf) @@ -117,6 +124,9 @@ object CommandLineInterface { case Nil => map case ("--db" | "--db-path") :: value :: tail => nextArg(map ++ Map("db-path" -> value), tail) + case "--skip-write" :: tail => + // A valueless boolean flag; it must match before the generic pair below swallows the next token. + nextArg(map ++ Map("skip-write" -> true), tail) case flag :: value :: tail if flag.startsWith("--") => // Generic `--flag value` pair; command providers translate these into positional args. nextArg(map ++ Map(flag.stripPrefix("--") -> value), tail) diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/data/write/SinkHandler.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/data/write/SinkHandler.scala index 9a61fa4e..10485a67 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/data/write/SinkHandler.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/data/write/SinkHandler.scala @@ -39,8 +39,14 @@ object SinkHandler { val fhirWriteProblemsAccum: CollectionAccumulator[FhirMappingResult] = spark.sparkContext.collectionAccumulator[FhirMappingResult](accumName) fhirWriteProblemsAccum.reset() - // Write the FHIR resources - resourceWriter.write(spark, mappedResults, fhirWriteProblemsAccum) + // Write the FHIR resources, unless the execution skips writing (the accumulator then stays empty, so the mapped + // resources are logged as numOfFhirResources with no failed writes) + if (mappingJobExecution.isWriteSkipped) + logger.info( + s"Skipping the write of the mapped resources for mappingTask '$mappingTaskName' of execution '${mappingJobExecution.id}'." + ) + else + resourceWriter.write(spark, mappedResults, fhirWriteProblemsAccum) logMappingJobResult( mappingJobExecution, mappingTaskName, diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/MappingJobLauncher.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/MappingJobLauncher.scala index de8383bc..e5ea14a9 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/MappingJobLauncher.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/MappingJobLauncher.scala @@ -69,6 +69,9 @@ class MappingJobLauncher(ignifyrEngine: IgnifyrEngine)(implicit ec: ExecutionCon * @param ignifyrDbFolderPath Ignifyr database folder (scheduling providers keep last-sync state there). * @param clearCheckpoints For streaming jobs, reset archiving offsets and delete the Spark * checkpoint directories so the streams start from scratch. + * @throws IllegalArgumentException if the execution skips writing but the job is streaming or scheduled: skipping + * the write is supported only for batch executions, since a stream would still + * advance its checkpoints and a schedule its last-sync time. */ def launch( mappingJob: FhirMappingJob, @@ -76,6 +79,13 @@ class MappingJobLauncher(ignifyrEngine: IgnifyrEngine)(implicit ec: ExecutionCon ignifyrDbFolderPath: String = IgnifyrConfig.engineConfig.ignifyrDbFolderPath, clearCheckpoints: Boolean = false ): MappingJobLaunch = { + if ( + mappingJobExecution.isWriteSkipped && + (mappingJob.sourceSettings.exists(_._2.asStream) || mappingJob.schedulingSettings.nonEmpty) + ) + throw new IllegalArgumentException( + s"Skipping the write is supported only for batch executions, but job '${mappingJob.id}' is streaming or scheduled." + ) if (mappingJob.sourceSettings.exists(_._2.asStream)) { if (clearCheckpoints) clearCheckpointDirectories(mappingJobExecution) val queryRegistrations = jobManager diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/log/ExecutionLogger.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/log/ExecutionLogger.scala index 7db57105..796ff06f 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/log/ExecutionLogger.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/log/ExecutionLogger.scala @@ -187,6 +187,7 @@ object ExecutionLogger { markerMap.put("projectId", mappingJobExecution.projectId) markerMap.put("executionId", mappingJobExecution.id) markerMap.put("mappingTaskName", mappingTaskName) + markerMap.put("isWriteSkipped", mappingJobExecution.isWriteSkipped) // log the chunk progress for batch jobs if (!mappingJobExecution.isStreamingJob) { markerMap.put("chunkProgress", s"0 / $numOfChunks") @@ -238,6 +239,7 @@ object ExecutionLogger { markerMap.put("projectId", mappingJobExecution.projectId) markerMap.put("executionId", mappingJobExecution.id) markerMap.put("mappingTaskName", mappingTaskName) + markerMap.put("isWriteSkipped", mappingJobExecution.isWriteSkipped) // log the batch progress for batch jobs if (!mappingJobExecution.isStreamingJob) { markerMap.put("chunkProgress", s"0 / 1") // chunk size is not determined yet diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/processing/FileStreamInputArchiver.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/processing/FileStreamInputArchiver.scala index 66c39590..4427bf81 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/processing/FileStreamInputArchiver.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/execution/processing/FileStreamInputArchiver.scala @@ -102,7 +102,12 @@ object FileStreamInputArchiver { // That means archiving works in best-effort mode. try { val archiveMode: ArchiveModes = execution.archiveMode - if (archiveMode != ArchiveModes.OFF) { + // An execution that skips writing must leave its input files in place: it wrote nothing, so they are not consumed + if (execution.isWriteSkipped && archiveMode != ArchiveModes.OFF) + logger.info( + s"Skipping archiving for job: ${execution.jobId}, execution: ${execution.id}, since its write is skipped" + ) + else if (archiveMode != ArchiveModes.OFF) { // get data folder path from data source settings val dataFolderPath = FileUtils.getPath(execution.fileSystemSourceDataFolderPath.get).toString diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/job/FhirMappingJobManager.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/job/FhirMappingJobManager.scala index 4da68961..ee6d3eee 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/job/FhirMappingJobManager.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/mapping/job/FhirMappingJobManager.scala @@ -90,7 +90,8 @@ class FhirMappingJobManager( timeRange: Option[(LocalDateTime, LocalDateTime)] = None ): Future[Unit] = { val sinkWriter = SinkWriterFactory.apply(sinkSettings) - sinkWriter.validate() + // The sink is not touched at all when the execution skips writing + if (!mappingJobExecution.isWriteSkipped) sinkWriter.validate() mappingJobExecution.mappingTasks.foldLeft(Future((): Unit)) { (f, task) => // Initial empty Future f.flatMap { _ => // Execute the Futures in the Sequence consecutively (not in parallel) // log the start of the FHIR mapping task execution @@ -212,7 +213,8 @@ class FhirMappingJobManager( identityServiceSettings: Option[IdentityServiceSettings] = None ): Future[Unit] = { val sinkWriter = SinkWriterFactory.apply(sinkSettings) - sinkWriter.validate() + // The sink is not touched at all when the execution skips writing + if (!mappingJobExecution.isWriteSkipped) sinkWriter.validate() readSourceAndExecuteTask( mappingJobExecution.jobId, diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobExecution.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobExecution.scala index afae0298..9f48204c 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobExecution.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobExecution.scala @@ -21,6 +21,9 @@ import java.util.regex.Pattern * @param archiveMode Archive mode of execution * @param saveErroneousRecords Whether to save erroneous records or not * @param isScheduledJob Whether the execution is scheduled or not + * @param isWriteSkipped Whether writing the mapped resources to the sink is skipped (a dry run: the + * mappings run and their results are logged, but nothing reaches the sink and + * the source files are not archived). Supported only for batch executions. */ case class FhirMappingJobExecution( id: String, @@ -32,7 +35,8 @@ case class FhirMappingJobExecution( fileSystemSourceDataFolderPath: Option[String], archiveMode: ArchiveModes, saveErroneousRecords: Boolean, - isScheduledJob: Boolean + isScheduledJob: Boolean, + isWriteSkipped: Boolean ) { /** @@ -144,6 +148,7 @@ object FhirMappingJobExecution { * @param job FHIR mapping job that includes this execution. * @param mappingTasks List of mapping tasks to be executed (as a subset of the mapping tasks defined in the job) * @param jobGroupIdOrStreamingQuery Keeps Spark job group id for batch jobs and StreamingQuery for streaming jobs + * @param skipWrite Whether to skip writing the mapped resources to the sink (see [[FhirMappingJobExecution.isWriteSkipped]]) * @return */ def apply( @@ -151,7 +156,8 @@ object FhirMappingJobExecution { projectId: String = "", job: FhirMappingJob, mappingTasks: Seq[FhirMappingTask] = Seq.empty, - jobGroupIdOrStreamingQuery: Option[Either[String, collection.mutable.Map[String, StreamingQuery]]] = None + jobGroupIdOrStreamingQuery: Option[Either[String, collection.mutable.Map[String, StreamingQuery]]] = None, + skipWrite: Boolean = false ): FhirMappingJobExecution = { // Configure properties related to the source settings of the job @@ -186,7 +192,8 @@ object FhirMappingJobExecution { fileSystemSourceDataFolderPath, archiveMode, saveErroneousRecords, - isScheduledJob + isScheduledJob, + skipWrite ) } } diff --git a/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobResult.scala b/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobResult.scala index 5f0a7d37..dd5cb10f 100644 --- a/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobResult.scala +++ b/ignifyr-engine/src/main/scala/io/ignifyr/engine/model/FhirMappingJobResult.scala @@ -66,7 +66,8 @@ case class FhirMappingJobResult( s"\t# of Invalid Rows: \t$numOfInvalids\n" + s"\t# of Not Mapped: \t$numOfNotMapped\n" + s"\t# of Failed writes:\t$numOfFailedWrites\n" + - s"\t# of Written FHIR resources:\t$numOfFhirResources" + (if (mappingJobExecution.isWriteSkipped) s"\t# of FHIR resources (write skipped):\t$numOfFhirResources" + else s"\t# of Written FHIR resources:\t$numOfFhirResources") } message } @@ -99,6 +100,7 @@ case class FhirMappingJobResult( markerMap.put("eventId", eventId) markerMap.put("isStreamingJob", mappingJobExecution.isStreamingJob) markerMap.put("isScheduledJob", mappingJobExecution.isScheduledJob) + markerMap.put("isWriteSkipped", mappingJobExecution.isWriteSkipped) // log the chunk progress for batch jobs if (!mappingJobExecution.isStreamingJob) { markerMap.put("chunkProgress", s"$completedNumOfChunks / $totalNumOfChunks") diff --git a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/cli/CommandLineInterfaceTest.scala b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/cli/CommandLineInterfaceTest.scala index 8e4f9ed8..81bcec9c 100644 --- a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/cli/CommandLineInterfaceTest.scala +++ b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/cli/CommandLineInterfaceTest.scala @@ -64,4 +64,18 @@ class CommandLineInterfaceTest extends AnyFlatSpec with Matchers { it should "let a later flag override an earlier one with the same name" in { parse("run", "--job", "first.json", "--job", "second.json") should contain("job" -> "second.json") } + + // --skip-write takes no value: read as a generic pair it would swallow the next token (here `--job`), + // and the job path would be lost. + it should "read --skip-write as a valueless flag anywhere on the line" in { + parse("run", "--skip-write", "--job", "jobs/patient.json") shouldBe + Map("command" -> "run", "skip-write" -> true, "job" -> "jobs/patient.json") + parse("run", "--job", "jobs/patient.json", "--skip-write") shouldBe + Map("command" -> "run", "skip-write" -> true, "job" -> "jobs/patient.json") + parse("--skip-write", "run") shouldBe Map("command" -> "run", "skip-write" -> true) + } + + it should "not set skip-write unless it is given" in { + parse("run", "--job", "jobs/patient.json") should not contain key("skip-write") + } } diff --git a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/data/SinkHandlerTest.scala b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/data/SinkHandlerTest.scala index be2101e0..91917776 100644 --- a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/data/SinkHandlerTest.scala +++ b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/data/SinkHandlerTest.scala @@ -25,11 +25,15 @@ class SinkHandlerTest extends AnyFlatSpec with Matchers { /** Captures the dataset it is handed instead of writing it anywhere. */ private class CapturingWriter extends BaseSinkWriter(FileSystemSinkSettings("./out", SinkContentTypes.NDJSON)) { var written: Seq[FhirMappingResult] = Seq.empty + var writeCalls: Int = 0 override def write( spark: SparkSession, df: Dataset[FhirMappingResult], problemsAccumulator: CollectionAccumulator[FhirMappingResult] - ): Unit = written = df.collect().toSeq + ): Unit = { + writeCalls += 1 + written = df.collect().toSeq + } override def validate(): Unit = () } @@ -104,6 +108,24 @@ class SinkHandlerTest extends AnyFlatSpec with Matchers { } it should "call the writer even when there is nothing to write" in { - write() shouldBe empty + import sparkSession.implicits._ + val writer = new CapturingWriter + SinkHandler.writeMappingResult(sparkSession, execution, "task-1", Seq.empty[FhirMappingResult].toDS(), writer) + writer.writeCalls shouldBe 1 + } + + // An execution that skips writing still maps and logs, but its mapped resources never reach the writer. + it should "not call the writer when the execution skips writing" in { + import sparkSession.implicits._ + val writer = new CapturingWriter + SinkHandler.writeMappingResult( + sparkSession, + execution.copy(isWriteSkipped = true), + "task-1", + Seq(result("row-1", mappedResource = Some("""{"resourceType":"Patient"}"""))).toDS(), + writer + ) + writer.writeCalls shouldBe 0 + writer.written shouldBe empty } } diff --git a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/FileStreamInputArchiverTest.scala b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/FileStreamInputArchiverTest.scala index 6912ba76..3eeadedc 100644 --- a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/FileStreamInputArchiverTest.scala +++ b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/FileStreamInputArchiverTest.scala @@ -181,6 +181,24 @@ class FileStreamInputArchiverTest extends AnyFlatSpec with Matchers { org.apache.commons.io.FileUtils.deleteDirectory(FileUtils.getPath(sourceFolderPath).toFile) } + "FileStreamInputArchiver" should "not apply deletion for a batch job that skips writing" in { + + // Create a test input file + val inputFile = initializeInputFiles(sourceFolderPath, inputFilePath) + + // Check whether input file exists + inputFile.exists() shouldBe true + + // Call archiving function for an execution that wrote nothing + FileStreamInputArchiver.applyArchivingOnBatchJob(testExecutionWithDelete.copy(isWriteSkipped = true)) + + // Check whether input file remains + inputFile.exists() shouldBe true + + // Clean test directories + org.apache.commons.io.FileUtils.deleteDirectory(FileUtils.getPath(sourceFolderPath).toFile) + } + "FileStreamInputArchiver" should "get input files" in { val mappingTaskName = "mocked_mappingTask_name" diff --git a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/MappingJobLauncherSkipWriteTest.scala b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/MappingJobLauncherSkipWriteTest.scala new file mode 100644 index 00000000..d600243c --- /dev/null +++ b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/execution/MappingJobLauncherSkipWriteTest.scala @@ -0,0 +1,45 @@ +package io.ignifyr.test.engine.execution + +import io.ignifyr.engine.IgnifyrEngine +import io.ignifyr.engine.execution.MappingJobLauncher +import io.ignifyr.engine.model._ +import org.mockito.MockitoSugar._ +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import scala.concurrent.ExecutionContext.Implicits.global + +/** + * Skipping the write is supported only for batch executions: a streaming execution would still advance its + * checkpoints and a scheduled one its last-sync time, so a later real run would silently skip the data the + * skipped run consumed. The launcher, the single dispatch point for the CLI and the server, refuses both + * before anything starts. + */ +class MappingJobLauncherSkipWriteTest extends AnyFlatSpec with Matchers { + + private val launcher = new MappingJobLauncher(mock[IgnifyrEngine]) + + private val sinkSettings = FileSystemSinkSettings("./out", SinkContentTypes.NDJSON) + + private def job(asStream: Boolean = false, scheduling: Option[BaseSchedulingSettings] = None): FhirMappingJob = + FhirMappingJob( + id = "job-1", + sourceSettings = Map("_" -> FileSystemSourceSettings("test", "test", "data", asStream = asStream)), + sinkSettings = sinkSettings, + mappings = Seq.empty, + schedulingSettings = scheduling + ) + + private def skippedExecution(mappingJob: FhirMappingJob): FhirMappingJobExecution = + FhirMappingJobExecution(job = mappingJob, skipWrite = true) + + "launch" should "refuse to skip the write of a streaming job" in { + val streamingJob = job(asStream = true) + an[IllegalArgumentException] should be thrownBy launcher.launch(streamingJob, skippedExecution(streamingJob)) + } + + it should "refuse to skip the write of a scheduled job" in { + val scheduledJob = job(scheduling = Some(SchedulingSettings("* * * * *"))) + an[IllegalArgumentException] should be thrownBy launcher.launch(scheduledJob, skippedExecution(scheduledJob)) + } +} diff --git a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/model/FhirMappingJobExecutionTest.scala b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/model/FhirMappingJobExecutionTest.scala index 69e548d3..e938f314 100644 --- a/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/model/FhirMappingJobExecutionTest.scala +++ b/ignifyr-engine/src/test/scala/io/ignifyr/test/engine/model/FhirMappingJobExecutionTest.scala @@ -1,6 +1,11 @@ package io.ignifyr.test.engine.model -import io.ignifyr.engine.model.{FhirMappingJob, FhirMappingJobExecution, FhirRepositorySinkSettings} +import io.ignifyr.engine.model.{ + FhirMappingJob, + FhirMappingJobExecution, + FhirMappingJobResult, + FhirRepositorySinkSettings +} import io.ignifyr.engine.util.SparkUtil import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -28,4 +33,24 @@ class FhirMappingJobExecutionTest extends AnyFlatSpec with Matchers { testExecution.getCommitDirectory(mappingTaskName) shouldBe SparkUtil.getCommitDirectoryPath(Paths.get(testExecution.getCheckpointDirectory(mappingTaskName))) } + + "FhirMappingJobExecution" should "write its results unless skipping the write is asked for" in { + testExecution.isWriteSkipped shouldBe false + FhirMappingJobExecution(job = testJob, skipWrite = true).isWriteSkipped shouldBe true + } + + // The Kibana dashboards read `isWriteSkipped` off every result event to tell a written execution from a skipped one + "FhirMappingJobResult" should "carry whether the write is skipped in its log marker" in { + def markerOf(execution: FhirMappingJobExecution) = + FhirMappingJobResult(execution, Some(mappingTaskName)).toMapMarker.getMap + markerOf(testExecution).get("isWriteSkipped") shouldBe false + markerOf(testExecution.copy(isWriteSkipped = true)).get("isWriteSkipped") shouldBe true + } + + it should "not report the resources of a skipped write as written" in { + val result = + FhirMappingJobResult(testExecution.copy(isWriteSkipped = true), Some(mappingTaskName), chunkResult = false) + result.toString should include("write skipped") + result.toString should not include "Written" + } }