stopper prosessering av meldinger når kafka-rapid skal stoppe

navikt · Nov 29, 2024 · 105481e · 105481e
1 parent cba36a8
commit 105481e
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 4 deletions.
diff --git a/rapids-and-rivers/src/main/kotlin/com/github/navikt/tbd_libs/rapids_and_rivers/KafkaRapid.kt b/rapids-and-rivers/src/main/kotlin/com/github/navikt/tbd_libs/rapids_and_rivers/KafkaRapid.kt
@@ -108,10 +108,13 @@ class KafkaRapid(
             .mapValues { it.value.minOf { it.offset() } }
             .toMutableMap()
         try {
-            records.onEach { record ->
-                onRecord(record)
-                currentPositions[TopicPartition(record.topic(), record.partition())] = record.offset() + 1
-            }
+            records
+                .onEach { record ->
+                    if (running.get()) {
+                        onRecord(record)
+                        currentPositions[TopicPartition(record.topic(), record.partition())] = record.offset() + 1
+                    }
+                }
         } catch (err: Exception) {
             log.info(
                 "due to an error during processing, positions are reset to each next message (after each record that was processed OK):" +

diff --git a/...vers/src/test/kotlin/com/github/navikt/tbd_libs/rapids_and_rivers/RapidIntegrationTest.kt b/...vers/src/test/kotlin/com/github/navikt/tbd_libs/rapids_and_rivers/RapidIntegrationTest.kt
@@ -24,6 +24,7 @@ import java.time.Duration
 import java.time.LocalDateTime
 import java.util.*
 import java.util.concurrent.TimeUnit.SECONDS
+import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.random.Random
 
 internal class RapidIntegrationTest {
@@ -84,6 +85,28 @@ internal class RapidIntegrationTest {
         assertDoesNotThrow { rapid.stop() }
     }
 
+    @Test
+    fun `stops while handling messages`() = rapidE2E {
+        val handlingMessages = AtomicBoolean(false)
+        rapid.register { message, context, metadata, metrics ->
+            handlingMessages.set(true)
+            runBlocking { delay(1000) }
+        }
+        repeat(100) { mainTopic.send("{}") }
+        await()
+            .atMost(10, SECONDS)
+            .until { handlingMessages.get() }
+
+        rapid.stop()
+
+        await()
+            .atMost(10, SECONDS)
+            .until {
+                runBlocking {  it.join() }
+                true
+            }
+    }
+
     @Test
     fun `should stop on errors`() {
         assertThrows<RuntimeException> {