tests

lib/stove-testing-e2e-kafka/src/main/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/intercepting/MessageSinkOps.kt

@@ -89,7 +89,6 @@ internal interface MessageSinkOps : MessageSinkPublishOps, CommonOps {
     }
 
     failedFunc.waitUntilCount(atLeastIn, times)
-    throwIfFailed(clazz, condition)
   }
 
   override fun dumpMessages(): String = "Sink so far:\n$store"

lib/stove-testing-e2e-kafka/src/test/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/setup/ProjectConfig.kt

@@ -69,6 +69,10 @@ class KafkaApplicationUnderTest : ApplicationUnderTest<Unit> {
 
 @ExperimentalKotest
 class ProjectConfig : AbstractProjectConfig() {
+  init {
+    stoveKafkaBridgePortDefault = "50052"
+  }
+
   override fun extensions(): List<Extension> = listOf(
     SystemEnvironmentProjectListener(STOVE_KAFKA_BRIDGE_PORT, stoveKafkaBridgePortDefault)
   )

lib/stove-testing-e2e-kafka/src/test/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/setup/example/StoveListener.kt

@@ -53,12 +53,11 @@ abstract class StoveListener(
             publisher.publishScope {
               offer(
                 ProducerRecord(
-                  // topic =
                   topicDefinition.retryTopic,
-                  // key =
+                  message.partition(),
                   message.key(),
-                  // value =
-                  message.value()
+                  message.value(),
+                  message.headers()
                 )
               )
             }
@@ -70,12 +69,11 @@ abstract class StoveListener(
         } else {
           logger.error("CONSUMER GOT an ERROR, retry limit exceeded: $message")
           val record = ProducerRecord<String, Any>(
-            // topic =
             topicDefinition.deadLetterTopic,
-            // key =
+            message.partition(),
             message.key(),
-            // value =
-            message.value()
+            message.value(),
+            message.headers()
           ).apply {
             headers().add("doNotFail", "true".toByteArray())
           }

lib/stove-testing-e2e-kafka/src/test/kotlin/com/trendyol/stove/testing/e2e/standalone/kafka/setup/example/consumers/ProductConsumer.kt

@@ -13,6 +13,6 @@ class ProductConsumer(
   override val topicDefinition: TopicDefinition = TopicDefinition("product", "product.retry", "product.error")
 
   override suspend fun listen(record: ConsumerRecord<String, String>) {
-    logger.info("Product consumed: ${record.value()}")
+    logger.info("Product consumed: ${record.value()} from topic: ${record.topic()} with key: ${record.key()}")
   }
 }