Stress testing

Author: nomisRev

.github/workflows/pr.yaml

@@ -23,7 +23,7 @@ jobs:
 
       - uses: gradle/gradle-build-action@v2
         with:
-          arguments: build --scan --full-stacktrace
+          arguments: build --scan --full-stacktrace -PstressTest=100
 
       - name: Bundle the build report
         if: failure()

build.gradle.kts

@@ -80,6 +80,10 @@ tasks {
 
   withType<Test>().configureEach {
     useJUnitPlatform()
+    maxParallelForks = (2 * Runtime.getRuntime().availableProcessors())
+    if (project.hasProperty("stressTest")) {
+      systemProperty("io.github.nomisrev.kafka.TEST_ITERATIONS", project.properties["stressTest"] ?: 100)
+    }
     testLogging {
       exceptionFormat = FULL
       events = setOf(SKIPPED, FAILED, STANDARD_ERROR)

src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt

@@ -49,6 +49,8 @@ import java.util.concurrent.TimeUnit
 import kotlin.test.assertEquals
 import kotlin.time.Duration.Companion.seconds
 
+private val testIterations: Int =
+  System.getProperties().getProperty("io.github.nomisrev.kafka.TEST_ITERATIONS")?.toIntOrNull() ?: 1
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 abstract class KafkaSpec {
@@ -62,7 +64,7 @@ abstract class KafkaSpec {
     fun destroy() {
       kafka.stop()
     }
-    
+
     @BeforeAll
     @JvmStatic
     fun setup() {
@@ -85,10 +87,10 @@ abstract class KafkaSpec {
         withEnv("KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND", "true")
         withReuse(true)
       }
-    
+
     fun KafkaReceiver(): KafkaReceiver<String, String> =
       KafkaReceiver(receiverSetting())
-    
+
     fun receiverSetting(): ReceiverSettings<String, String> =
       ReceiverSettings(
         bootstrapServers = kafka.bootstrapServers,
@@ -168,15 +170,17 @@ abstract class KafkaSpec {
     partitions: Int = 4,
     replicationFactor: Short = 1,
     test: suspend TopicTestScope.(NewTopic) -> Unit
-  ): Unit = runTest {
-    val topic = NewTopic(nextTopicName(), partitions, replicationFactor).configs(topicConfig)
-    admin {
-      createTopic(topic)
-      try {
-        TopicTestScope(topic, this@runTest).test(topic)
-      } finally {
-        topic.shouldBeEmpty()
-        deleteTopic(topic.name())
+  ): Unit = repeat(testIterations) {
+    runTest {
+      val topic = NewTopic(nextTopicName(), partitions, replicationFactor).configs(topicConfig)
+      admin {
+        createTopic(topic)
+        try {
+          TopicTestScope(topic, this@runTest).test(topic)
+        } finally {
+          topic.shouldBeEmpty()
+          deleteTopic(topic.name())
+        }
       }
     }
   }
@@ -295,6 +299,9 @@ abstract class KafkaSpec {
     {
       val producer = KafkaProducer(it.properties(), it.keySerializer, it.valueSerializer)
       object : Producer<String, String> {
+        override fun clientInstanceId(p0: Duration?): Uuid =
+          producer.clientInstanceId(p0)
+
         override fun close() {}
 
         override fun close(timeout: Duration?) {}