fix stovelistener for testing

Trendyol · May 1, 2024 · 1bdb6d1 · 1bdb6d1
1 parent a67a121
commit 1bdb6d1
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/...est/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/setup/example/StoveListener.kt b/...est/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/setup/example/StoveListener.kt
@@ -25,7 +25,7 @@ abstract class StoveListener(
   suspend fun start() {
     consumer.subscribe(listOf(topicDefinition.topic, topicDefinition.retryTopic, topicDefinition.deadLetterTopic))
     consuming = GlobalScope.launch {
-      while (!consuming.isCancelled) {
+      while (!this.isActive) {
         consumer
           .poll(Duration.ofMillis(100))
           .forEach { message ->
@@ -84,7 +84,7 @@ abstract class StoveListener(
 
   abstract suspend fun listen(record: ConsumerRecord<String, String>)
 
-  override fun close() {
-    Try { consuming.cancel() }
+  override fun close(): Unit = runBlocking {
+    Try { consuming.cancelAndJoin() }
   }
 }