Do not include records per seconds if there are not many records.

AbsaOSS · Jan 11, 2024 · fd79310 · fd79310
1 parent 83204ac
commit fd79310
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 6 deletions.
diff --git a/...c/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala b/...c/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala
@@ -22,7 +22,7 @@ import za.co.absa.pramen.api.notification._
 import za.co.absa.pramen.core.config.Keys.TIMEZONE
 import za.co.absa.pramen.core.exceptions.{CmdFailedException, ProcessFailedException}
 import za.co.absa.pramen.core.notify.message._
-import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml.MIN_RPS_JOB_DURATION_SECONDS
+import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml.{MIN_RPS_JOB_DURATION_SECONDS, MIN_RPS_RECORDS}
 import za.co.absa.pramen.core.pipeline.TaskRunReason
 import za.co.absa.pramen.core.runner.task.RunStatus._
 import za.co.absa.pramen.core.runner.task.{NotificationFailure, RunStatus, TaskResult}
@@ -36,6 +36,7 @@ import scala.collection.mutable.ListBuffer
 
 object PipelineNotificationBuilderHtml {
   val MIN_RPS_JOB_DURATION_SECONDS = 60
+  val MIN_RPS_RECORDS = 1000
 }
 
 class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNotificationBuilder {
@@ -450,7 +451,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     task.runInfo match {
       case Some(runInfo) =>
         val jobDuration = Duration.between(runInfo.started, runInfo.finished).getSeconds
-        if (jobDuration > MIN_RPS_JOB_DURATION_SECONDS && recordCount > 0L) {
+        if (jobDuration > MIN_RPS_JOB_DURATION_SECONDS && recordCount >= MIN_RPS_RECORDS) {
           val throughput = recordCount / jobDuration
 
           throughput match {

diff --git a/pramen/core/src/test/resources/test/notify/test_pipeline_email_body_complex.txt b/pramen/core/src/test/resources/test/notify/test_pipeline_email_body_complex.txt
@@ -122,10 +122,10 @@
 <td>table_out</td>
 <td>db.table1</td>
 <td style="text-align:center">2022-02-18</td>
-<td style="text-align:right">200 (+100)</td>
+<td style="text-align:right">20000 (+10000)</td>
 <td style="text-align:center">01:14:04</td>
-<td style="text-align:right">1000 B</td>
-<td class="tdwarn" style="text-align:right">0</td>
+<td style="text-align:right">97 KiB</td>
+<td class="tdwarn" style="text-align:right">4</td>
 <td style="text-align:center">1970-01-01 03:34 +0200</td>
 <td class="tdwarn" style="text-align:center">Warning</td>
 <td>Test warning</td></tr></tbody>

diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TestPrototypes.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TestPrototypes.scala
@@ -24,7 +24,7 @@ object TestPrototypes {
   val runStatusSuccess: RunStatus = RunStatus.Succeeded(Some(100), 200, Some(1000), TaskRunReason.New, Seq.empty, Seq.empty, Seq.empty, Seq.empty)
 
   val runStatusWarning: RunStatus = RunStatus.Succeeded(
-    Some(100), 200, Some(1000), TaskRunReason.New, Seq("file1.txt", "file1.ctl"),
+    Some(10000), 20000, Some(100000), TaskRunReason.New, Seq("file1.txt", "file1.ctl"),
     Seq("file1.csv", "file2.csv"), Seq("`db`.`table1`"), Seq("Test warning")
   )