when consumer is closed, give a signal

navikt · Nov 2, 2023 · a85e87a · a85e87a
1 parent 4c6a3a6
commit a85e87a
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/hm-rapids-and-rivers-v2-core/src/main/kotlin/no/nav/helse/rapids_rivers/KafkaRapid.kt b/hm-rapids-and-rivers-v2-core/src/main/kotlin/no/nav/helse/rapids_rivers/KafkaRapid.kt
@@ -28,6 +28,7 @@ class KafkaRapid(
     private val running = AtomicBoolean(Stopped)
     private val ready = AtomicBoolean(false)
     private val producerClosed = AtomicBoolean(false)
+    private val consumerClosed = AtomicBoolean(false)
 
     private val stringDeserializer = StringDeserializer()
     private val stringSerializer = StringSerializer()
@@ -58,7 +59,7 @@ class KafkaRapid(
 
     fun isRunning() = running.get()
     fun isReady() = isRunning() && ready.get()
-    fun isConsumerClosed() = !isRunning()
+    fun isConsumerClosed() = consumerClosed.get()
     fun isProducerClosed() = producerClosed.get()
 
     override fun publish(message: String) {
@@ -231,6 +232,7 @@ class KafkaRapid(
         } else {
             log.info("stopped consuming messages after receiving stop signal")
         }
+        consumerClosed.set(true)
         tryAndLog(consumer::unsubscribe)
         tryAndLog(consumer::close)
     }