use import-process-bridge reports

Signed-off-by: Sebastian Schüpbach <sebastian.schuepbach@unibas.ch>

k8s-manifests/job.yml

@@ -35,9 +35,9 @@ spec:
             - name: CLIENT_ID
               value: import-process-delete
             - name: TOPIC_IN
-              value: import-process-reporting
+              value: postprocessing-reporting
             - name: TOPIC_OUT
-              value: import-process-test
+              value: import-process-test-delete
             - name: POLL_TIMEOUT
               value: "60000"
       restartPolicy: Never

src/main/scala/ch/memobase/App.scala

@@ -21,7 +21,6 @@ package ch.memobase
 import ch.memobase.models.{DeleteMessage, ParserIgnore, Report}
 import org.apache.logging.log4j.scala.Logging
 
-import scala.collection.mutable
 import scala.util.{Failure, Success}
 
 object App
@@ -36,7 +35,6 @@ object App
   parse(args) match {
     case Some((sessionId, filters, dryRun)) =>
       try {
-        val recordSetIds: mutable.HashSet[String] = mutable.HashSet()
         val matches = Iterator.continually(poll)
           .takeWhile(_.nonEmpty)
           .flatten
@@ -50,10 +48,6 @@ object App
                 logger.info("Match found")
                 List(DeleteMessage(rep, sessionId))
               case Success(r) =>
-                if (!recordSetIds.contains(r.recordSetId)) {
-                  recordSetIds += r.recordSetId
-                  logger.info(r.recordSetId)
-                }
                 List()
               case Failure(_: ParserIgnore) =>
                 logger.debug("Ignoring irrelevant message")

src/main/scala/ch/memobase/models/Report.scala

@@ -41,12 +41,12 @@ object Report {
   def apply(consumerRecord: ConsumerRecord[String, String]): Try[Report] = Try {
     val json = Try(ujson.read(consumerRecord.value())).getOrElse(throw new ParserException("JSON is not valid!"))
     val step = Try(json.obj("step").str).getOrElse(throw new ParserException("No `step` field in JSON obj"))
-    if (step != "normalization-service") {
+    if (step != "import-process-bridge") {
       throw new ParserIgnore("No normalization-service message")
     }
     val status = Try(json.obj("status").str).getOrElse(throw new ParserException("No `status` field in JSON obj"))
-    if (status != "SUCCESS") {
-      throw new ParserIgnore("No SUCCESS message")
+    if (status == "FATAL") {
+      throw new ParserIgnore("FATAL message: No record forwarded, therefore ignoring report")
     }
     val id = Try(json.obj("id").str).getOrElse(throw new ParserException("No `id` field in JSON obj"))
     val timestampString = Try(json.obj("timestamp").str).getOrElse(throw new ParserException("No `timestamp` field in JSON obj"))