Acknowledge offset

navikt · Dec 19, 2024 · e158fc1 · e158fc1
1 parent a37fd71
commit e158fc1
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 5 deletions.
diff --git a/ebms-provider/src/main/kotlin/no/nav/emottak/ebms/App.kt b/ebms-provider/src/main/kotlin/no/nav/emottak/ebms/App.kt
@@ -30,7 +30,6 @@ import io.micrometer.prometheus.PrometheusConfig
 import io.micrometer.prometheus.PrometheusMeterRegistry
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.awaitCancellation
-import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.launch
 import net.logstash.logback.marker.Markers
@@ -95,8 +94,12 @@ fun main() = SuspendApp {
                     KafkaReceiver(receiverSettings)
                         .receive(kafkaConfig.incomingSignalTopic)
                         .take(10)
-                        .map { it.key() to it.value() }
-                        .collect(signalProcessor::processSignal)
+                        .collect { record ->
+                            signalProcessor.processSignal(record.key(), record.value())
+                            record.offset.acknowledge().also {
+                                log.debug("Acknowledged topic offset ${record.offset()}")
+                            }
+                        }
                 }
             }
 

diff --git a/ebms-provider/src/main/kotlin/no/nav/emottak/ebms/processing/SignalProcessor.kt b/ebms-provider/src/main/kotlin/no/nav/emottak/ebms/processing/SignalProcessor.kt
@@ -4,7 +4,7 @@ import no.nav.emottak.ebms.log
 
 class SignalProcessor {
 
-    fun processSignal(signal: Pair<String, ByteArray>) {
-        log.info("Got signal with reference <${signal.first}> and content: ${String(signal.second)}")
+    fun processSignal(reference: String, content: ByteArray) {
+        log.info("Got signal with reference <$reference> and content: ${String(content)}")
     }
 }