Improve reporting

Update dependency.

build.gradle

@@ -33,7 +33,7 @@ ext {
 dependencies {
     implementation 'ch.memobase:memobase-kafka-utils:0.2.3'
     implementation 'org.memobase:memobase-service-utilities:2.0.5'
-    implementation 'ch.memobase:mapper-service-configuration:1.0.2'
+    implementation 'ch.memobase:mapper-service-configuration:1.0.3'
     // Logging Framework
     implementation "org.apache.logging.log4j:log4j-api:${log4jV}"
     implementation "org.apache.logging.log4j:log4j-core:${log4jV}"

src/main/kotlin/KafkaTopology.kt

@@ -45,7 +45,7 @@ class KafkaTopology(
     private val settings: SettingsLoader
 ) {
 
-    private val log = LogManager.getLogger("KafkaTopology")
+    private val log = LogManager.getLogger("MappingProcessor")
     private val reportTopic = settings.processReportTopic
     private val klaxon = Klaxon()
 
@@ -75,7 +75,7 @@ class KafkaTopology(
             configJoiner.join(stream, configStream)
 
         val handledStream = joinedStream
-            .mapValues { readOnlyKey, value -> handleExceptions(value) }
+            .mapValues { value -> handleExceptions(value) }
             .branch(
                 Predicate { _, value -> value.third != "" },
                 Predicate { _, _ -> true }
@@ -173,7 +173,7 @@ class KafkaTopology(
         val noRecordId = hasRecordId[1]
 
         noRecordId
-            .mapValues { key, _ -> Report(key, ReportStatus.fatal, "No record id found for record $key.", Service.step).toJson() }
+            .mapValues { key, value -> Report(key, ReportStatus.fatal, value.errorMessage, Service.step).toJson() }
             .to(reportTopic)
 
         return hasRecordId[0]
@@ -190,11 +190,11 @@ class KafkaTopology(
         val noRecordTypeValue = hasRecordTypeValue[1]
 
         noRecordTypeValue
-            .mapValues { key, _ ->
+            .mapValues { key, value ->
                 Report(
                     key,
                     ReportStatus.fatal,
-                    "No correct record type found for record $key.",
+                    value.errorMessage,
                     Service.step
                 ).toJson()
             }
@@ -209,7 +209,7 @@ class KafkaTopology(
             result.first,
             Pair(
                 result.second,
-                Report(result.first, ReportStatus.success, "Successfully mapped record with id ${result.first}.", Service.step)
+                Report(result.first, ReportStatus.success, "", Service.step)
             )
         )
     }
@@ -230,7 +230,7 @@ class KafkaTopology(
                 Triple(parsedSource, mapperConfiguration.get(), null)
             } else {
                 log.error("Parsed source is empty: ${value.first}.")
-                Triple(null, null, Report(key, ReportStatus.fatal, "Found empty source document.", Service.step))
+                Triple(null, null, Report(key, ReportStatus.fatal, "Could not parse source document.", Service.step))
             }
         } catch (ex: InvalidMappingException) {
             log.error(ex.localizedMessage)

src/test/kotlin/IntegrationTests.kt

@@ -195,7 +195,7 @@ class IntegrationTests {
             Report(
                 "https://memobase.ch/record/test-record-set-id-ID_1",
                 "SUCCESS",
-                "Successfully mapped record with id https://memobase.ch/record/test-record-set-id-ID_1.",
+                "",
                 Service.step
             )
         )/*,