fix: add message key value while producing messages to kafka topic

1 files changed, 4 insertions, 3 deletions

diff --git a/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt b/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
index 374c0af..9520caa 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/output/KafkaMtbFileSender.kt
@@ -25,15 +25,16 @@ import org.slf4j.LoggerFactory
 import org.springframework.kafka.core.KafkaTemplate
 
 class KafkaMtbFileSender(
-    private val kafkaTemplate: KafkaTemplate<String, String>,
-    private val objectMapper: ObjectMapper
+        private val kafkaTemplate: KafkaTemplate<String, String>,
+        private val objectMapper: ObjectMapper
 ) : MtbFileSender {
 
     private val logger = LoggerFactory.getLogger(KafkaMtbFileSender::class.java)
 
     override fun send(mtbFile: MtbFile): MtbFileSender.Response {
         return try {
-            kafkaTemplate.sendDefault(objectMapper.writeValueAsString(mtbFile))
+            kafkaTemplate.sendDefault(String.format("{\"pid\": %s, \"eid\": %s}", mtbFile.patient.id,
+                    mtbFile.episode.id), objectMapper.writeValueAsString(mtbFile))
             logger.debug("Sent file via KafkaMtbFileSender")
             MtbFileSender.Response(MtbFileSender.ResponseStatus.UNKNOWN)
         } catch (e: Exception) {