#6 [SV] - Master dataset processing pipeline, updated the stats outpu…

…t to process stats for all events by removing the event itself
Sunbird-Obsrv · Apr 10, 2023 · 1e1e262 · 1e1e262
1 parent 8a6e896
commit 1e1e262
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 4 deletions.
diff --git a/...ssor/src/main/scala/org/sunbird/obsrv/pipeline/function/MasterDataProcessorFunction.scala b/...ssor/src/main/scala/org/sunbird/obsrv/pipeline/function/MasterDataProcessorFunction.scala
@@ -64,6 +64,9 @@ class MasterDataProcessorFunction(config: MasterDataProcessorConfig) extends Win
     metrics.incCounter(datasetId, config.successUpdateCount, result._2)
     metrics.incCounter(datasetId, config.successEventCount, eventsList.size.toLong)
 
-    context.output(config.successTag(), markComplete(eventsList.head))
+    eventsList.foreach(event => {
+      event.remove(config.CONST_EVENT)
+      context.output(config.successTag(), markComplete(event))
+    })
   }
 }
diff --git a/...sor/src/test/scala/org/sunbird/obsrv/pipeline/MasterDataProcessorStreamTaskTestSpec.scala b/...sor/src/test/scala/org/sunbird/obsrv/pipeline/MasterDataProcessorStreamTaskTestSpec.scala
@@ -88,9 +88,8 @@ class MasterDataProcessorStreamTaskTestSpec extends BaseSpecWithDatasetRegistry
       env.execute(masterDataConfig.jobName)
     }
 
-    val input = EmbeddedKafka.consumeNumberMessagesFrom[String](config.getString("kafka.stats.topic"), 3, timeout = 30.seconds)
-    //input.foreach(Console.println("Event:", _))
-    input.size should be (3)
+    val input = EmbeddedKafka.consumeNumberMessagesFrom[String](config.getString("kafka.stats.topic"), 4, timeout = 30.seconds)
+    input.size should be (4)
 
     val mutableMetricsMap = mutable.Map[String, Long]();
     BaseMetricsReporter.gaugeMetrics.toMap.mapValues(f => f.getValue()).map(f => mutableMetricsMap.put(f._1, f._2))