MqPlugin: fix KafkaTester + add support for sending params

Author: Adven27

README.md

@@ -21,15 +21,15 @@
 
 ```groovy
 // Typical microservices setup (Web API + DB + MQ) testing:
-testImplementation "io.github.adven27:exam-ms:5.0.1"
+testImplementation "io.github.adven27:exam-ms:5.1.0"
 
 //same as:
-//testImplementation "io.github.adven27:exam-ws:5.0.1"
-//testImplementation "io.github.adven27:exam-db:5.0.1"
-//testImplementation "io.github.adven27:exam-mq:5.0.1"
+//testImplementation "io.github.adven27:exam-ws:5.1.0"
+//testImplementation "io.github.adven27:exam-db:5.1.0"
+//testImplementation "io.github.adven27:exam-mq:5.1.0"
 
 // UI testing
-testImplementation "io.github.adven27:exam-ui:5.0.1"
+testImplementation "io.github.adven27:exam-ui:5.1.0"
 ```
 ### 2) Use
 

build.gradle

@@ -5,7 +5,7 @@ buildscript {
     ext.restAssured_version = '4.3.3'
     ext.wiremock_version = '2.27.2'
     ext.klogging_version = '2.0.6'
-    ext.libVersion = '5.0.1'
+    ext.libVersion = '5.1.0'
     repositories {
         mavenCentral()
     }

exam-core/src/main/java/io/github/adven27/concordion/extensions/exam/core/ExamExtension.kt

@@ -11,6 +11,8 @@ import io.github.adven27.concordion.extensions.exam.core.utils.DateWithin
 import io.github.adven27.concordion.extensions.exam.core.utils.HANDLEBARS
 import io.github.adven27.concordion.extensions.exam.core.utils.XMLDateWithin
 import io.github.adven27.concordion.extensions.exam.core.utils.equalToXml
+import io.github.adven27.concordion.extensions.exam.core.utils.prettyJson
+import io.github.adven27.concordion.extensions.exam.core.utils.prettyXml
 import mu.KotlinLogging
 import net.javacrumbs.jsonunit.JsonAssert
 import net.javacrumbs.jsonunit.JsonAssert.`when`
@@ -169,6 +171,11 @@ class ExamExtension constructor(private vararg var plugins: ExamPlugin) : Concor
             .withMatcher("formattedAndWithin", DateWithin.param())
             .withMatcher("formattedAndWithinNow", DateWithin.now())
             .withMatcher("xmlDateWithinNow", XMLDateWithin())
+
+        @JvmStatic
+        fun prettyXml(text: String) = text.prettyXml()
+        @JvmStatic
+        fun prettyJson(text: String) = text.prettyJson()
     }
 }
 

exam-core/src/main/java/io/github/adven27/concordion/extensions/exam/core/html/Html.kt

@@ -95,11 +95,6 @@ class Html(val el: Element) {
         return this
     }
 
-    infix fun aboveF(html: Html): Html {
-        el.parentElement.prependChild(html.el)
-        return this
-    }
-
     infix fun below(html: Html): Html {
         el.appendSister(html.el)
         return this
@@ -116,7 +111,6 @@ class Html(val el: Element) {
     }
 
     fun takeAwayAttr(name: String, def: String): String = takeAwayAttr(name) ?: def
-
     fun takeAwayAttr(attrName: String, def: String, eval: Evaluator? = null) = takeAwayAttr(attrName, eval) ?: def
 
     fun el() = el

exam-core/src/main/java/io/github/adven27/concordion/extensions/exam/core/utils/CheckUtils.kt

@@ -58,7 +58,8 @@ fun Document.prettyXml(): String {
 
 class InvalidXml(t: Throwable) : RuntimeException(t)
 
-fun Html.content(eval: Evaluator) = this.attr("from")?.findResource(eval)?.readText() ?: this.text().trim()
+fun Html.content(from: String?, eval: Evaluator) = from?.findResource(eval)?.readText() ?: this.text().trim()
+fun Html.content(eval: Evaluator) = content(this.attr("from"), eval)
 fun String.findResource(eval: Evaluator) = ExamExtension::class.java.getResource(eval.resolveJson(this))
     ?: throw FileNotFoundException("File not found: $this")
 

exam-mq-ibmmq/src/main/java/io/github/adven27/concordion/extensions/exam/mq/ibmmq/IbmMQTester.kt

@@ -33,11 +33,11 @@ open class IbmMQBrowseAndSendTester @JvmOverloads constructor(
 
     protected lateinit var producer: MessageProducer
 
-    override fun send(message: String, headers: Map<String, String>) {
+    override fun send(message: MqTester.Message, params: Map<String, String>) {
         logger.info("Sending to {}...", config.queue)
         producer.send(
-            sendConverter.apply(MqTester.Message(message, headers) to session).apply {
-                jmsCorrelationID = headers["jmsCorrelationID"]
+            sendConverter.apply(message to session).apply {
+                jmsCorrelationID = message.headers["jmsCorrelationID"]
             }
         )
         logger.info("Sent to {}\n{}", config.queue, message)
@@ -63,11 +63,11 @@ open class IbmMQConsumeAndSendTester @JvmOverloads constructor(
 
     protected lateinit var producer: MessageProducer
 
-    override fun send(message: String, headers: Map<String, String>) {
+    override fun send(message: MqTester.Message, params: Map<String, String>) {
         logger.info("Sending to {}...", config.queue)
         producer.send(
-            sendConverter.apply(MqTester.Message(message, headers) to session).apply {
-                jmsCorrelationID = headers["jmsCorrelationID"]
+            sendConverter.apply(message to session).apply {
+                jmsCorrelationID = message.headers["jmsCorrelationID"]
             }
         )
         logger.info("Sent to {}\n{}", config.queue, message)

exam-mq-kafka/src/main/java/io/github/adven27/concordion/extensions/exam/mq/kafka/KafkaTester.kt

@@ -4,112 +4,117 @@ import io.github.adven27.concordion.extensions.exam.mq.MqTester
 import mu.KLogging
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.OffsetSpec
+import org.apache.kafka.clients.admin.RecordsToDelete.beforeOffset
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.clients.consumer.OffsetAndMetadata
 import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.clients.producer.ProducerRecord
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.serialization.LongDeserializer
-import org.apache.kafka.common.serialization.LongSerializer
+import org.apache.kafka.common.header.internals.RecordHeader
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.serialization.StringSerializer
 import java.time.Duration
 import java.time.Duration.ofMillis
 import java.time.Duration.ofSeconds
-import java.util.Properties
 import java.util.concurrent.TimeUnit
 
 @Suppress("unused")
-open class KafkaTester @JvmOverloads constructor(
-    protected val bootstrapServers: String,
-    protected val topic: String,
-    protected val properties: Properties = DEFAULT_PROPERTIES,
-    protected val pollTimeout: Duration = ofMillis(POLL_MILLIS),
-    protected val partitionHeader: String = "partition"
-) : MqTester {
-    protected lateinit var producer: KafkaProducer<Long, String>
-    protected lateinit var consumer: KafkaConsumer<Long, String>
-
-    override fun purge() = logger.info("Purging topic {}...", topic).also {
-        consumer.poll(ofMillis(POLL_MILLIS))
-        logger.info("Topic {} is purged", topic)
-    }
-
-    override fun receive(): List<MqTester.Message> = logger.info("Reading from {}", topic).let {
-        consumer.poll(pollTimeout).apply { consumer.commitAsync() }.map { MqTester.Message(it.value()) }
-    }
-
-    override fun send(message: String, headers: Map<String, String>) = logger.info("Sending to {}...", topic).also {
-        producer.send(record(message, partitionFrom(headers))).get().apply {
-            logger.info(
-                "Sent to topic {} and partition {} with offset {}:\n{}", topic(), partition(), offset(), message
-            )
-        }
-    }
-
-    private fun partitionFrom(headers: Map<String, String>) = headers[partitionHeader]?.toInt()
-
-    private fun record(value: String, partition: Int?): ProducerRecord<Long, String> =
-        ProducerRecord(topic, partition, null, value)
+open class KafkaConsumeAndSendTester @JvmOverloads constructor(
+    sutConsumerGroup: String,
+    bootstrapServers: String,
+    topic: String,
+    properties: MutableMap<String, Any?> = (DEFAULT_PRODUCER_CONFIG + DEFAULT_CONSUMER_CONFIG).toMutableMap(),
+    pollTimeout: Duration = ofMillis(POLL_MILLIS),
+    accumulateOnRetries: Boolean = false
+) : KafkaConsumeOnlyTester(sutConsumerGroup, bootstrapServers, topic, properties, pollTimeout, accumulateOnRetries) {
+    protected lateinit var producer: KafkaProducer<String, String>
 
     override fun start() {
+        properties[ProducerConfig.CLIENT_ID_CONFIG] = "kafka-tester-$topic"
         properties[ProducerConfig.BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
-        producer = KafkaProducer<Long, String>(properties)
-        consumer = KafkaConsumer<Long, String>(properties).apply { subscribe(listOf(topic)) }
-        logger.info("KafkaTester started with properties:\n{}", properties)
+        producer = KafkaProducer<String, String>(properties)
+        super.start()
     }
 
     override fun stop() {
         producer.close(ofSeconds(4))
-        consumer.close(ofSeconds(4))
+        super.stop()
     }
 
+    override fun send(message: MqTester.Message, params: Map<String, String>) =
+        logger.info("Sending to {}...", topic).also {
+            producer.send(record(message, partitionFrom(params), keyFrom(params))).get().apply {
+                logger.info(
+                    "Sent to topic {} and partition {} with offset {}:\n{}", topic(), partition(), offset(), message
+                )
+            }
+        }
+
+    private fun record(message: MqTester.Message, partition: Int?, key: String?) = ProducerRecord(
+        topic, partition, key, message.body, message.headers.map { RecordHeader(it.key, it.value.toByteArray()) }
+    )
+
     companion object : KLogging() {
+        private const val POLL_MILLIS: Long = 1500
+        private const val FETCH_CONSUMER_GROUP_OFFSETS_TIMEOUT: Long = 10
+        private const val PARAM_PARTITION = "partition"
+        private const val PARAM_KEY = "key"
 
         @JvmField
-        val DEFAULT_PROPERTIES = Properties().apply {
-            put(ProducerConfig.CLIENT_ID_CONFIG, "kafka-tester")
-            put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer::class.java.name)
-            put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
-            put(ConsumerConfig.GROUP_ID_CONFIG, "kafka-tester")
-            put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer::class.java.name)
-            put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
-            put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
-            put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
-        }
+        val DEFAULT_PRODUCER_CONFIG: Map<String, String?> = mapOf(
+            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java.name,
+            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java.name,
+        )
 
-        protected const val POLL_MILLIS: Long = 1500
+        private fun partitionFrom(headers: Map<String, String>) = headers[PARAM_PARTITION]?.toInt()
+        private fun keyFrom(headers: Map<String, String>) = headers[PARAM_KEY]
     }
 }
 
-@Suppress("unused")
-open class KafkaMultiPartitionTester @JvmOverloads constructor(
-    private val sutConsumerGroup: String,
-    bootstrapServers: String,
-    topic: String,
-    properties: Properties = DEFAULT_PROPERTIES,
-    pollTimeout: Duration = ofMillis(POLL_MILLIS),
-    partitionHeader: String = "partition"
-) : KafkaTester(bootstrapServers, topic, properties, pollTimeout, partitionHeader) {
-    private var adminClient: AdminClient? = null
+@Suppress("unused", "TooManyFunctions")
+open class KafkaConsumeOnlyTester @JvmOverloads constructor(
+    protected val sutConsumerGroup: String,
+    protected val bootstrapServers: String,
+    protected val topic: String,
+    protected val properties: MutableMap<String, Any?> = DEFAULT_CONSUMER_CONFIG.toMutableMap(),
+    protected val pollTimeout: Duration = ofMillis(POLL_MILLIS),
+    protected val accumulateOnRetries: Boolean = false
+) : MqTester {
+    protected lateinit var consumer: KafkaConsumer<String, String>
+    protected lateinit var adminClient: AdminClient
+
+    override fun accumulateOnRetries() = accumulateOnRetries
 
     override fun start() {
-        super.start()
+        properties[ConsumerConfig.GROUP_ID_CONFIG] = "kafka-tester-$topic"
+        consumer = KafkaConsumer<String, String>(properties).apply { subscribe(listOf(topic)) }
         adminClient = AdminClient.create(properties)
+        logger.info("KafkaTester started with properties:\n{}", properties)
     }
 
-    override fun accumulateOnRetries() = false
+    override fun stop() {
+        consumer.close(ofSeconds(4))
+    }
+
+    override fun purge() = logger.info("Purging topic {}...", topic).also {
+        adminClient.deleteRecords(sutOffsets().map { it.key to beforeOffset(it.value.offset()) }.toMap())
+        logger.info("Topic {} is purged", topic)
+    }
 
     override fun receive(): List<MqTester.Message> = consumer.apply { seekTo(sutOffsets()) }.consume()
 
+    override fun send(message: MqTester.Message, params: Map<String, String>) {
+        throw UnsupportedOperationException("$javaClass doesn't support sending messages")
+    }
+
     private fun sutOffsets(): Map<TopicPartition, OffsetAndMetadata> =
-        adminClient!!.listConsumerGroupOffsets(sutConsumerGroup)
+        adminClient.listConsumerGroupOffsets(sutConsumerGroup)
             .partitionsToOffsetAndMetadata()[FETCH_CONSUMER_GROUP_OFFSETS_TIMEOUT, TimeUnit.SECONDS]
             .apply { logger.info("SUT offsets: {}", this) }
 
-    private fun KafkaConsumer<Long, String>.seekTo(offsets: Map<TopicPartition, OffsetAndMetadata>) {
+    private fun KafkaConsumer<String, String>.seekTo(offsets: Map<TopicPartition, OffsetAndMetadata>) {
         if (offsets.isEmpty()) {
             seekToBeginning()
         } else {
@@ -122,26 +127,35 @@ open class KafkaMultiPartitionTester @JvmOverloads constructor(
     }
 
     private fun endOf(p: TopicPartition): Long =
-        adminClient!!.listOffsets(mapOf(p to OffsetSpec.latest())).all().get()[p]?.offset() ?: 0L
+        adminClient.listOffsets(mapOf(p to OffsetSpec.latest())).all().get()[p]?.offset() ?: 0L
 
-    private fun KafkaConsumer<Long, String>.seekToBeginning() {
+    private fun KafkaConsumer<String, String>.seekToBeginning() {
         // At this point, there is no heartbeat from consumer and seek() wont work... So call poll() first
         poll(pollTimeout)
         seekToBeginning(assignment())
     }
 
-    private fun KafkaConsumer<Long, String>.seekTo(pointer: Long, p: TopicPartition) {
+    private fun KafkaConsumer<String, String>.seekTo(pointer: Long, p: TopicPartition) {
         // At this point, there is no heartbeat from consumer and seek() wont work... So call poll() first
         poll(pollTimeout)
         seek(p, pointer)
     }
 
-    private fun KafkaConsumer<Long, String>.consume(): List<MqTester.Message> = logger.info("Consuming events...").let {
-        poll(pollTimeout).apply { commitAsync() }.map { MqTester.Message(it.value()) }
-    }
+    private fun KafkaConsumer<String, String>.consume(): List<MqTester.Message> =
+        logger.info("Consuming events...").let {
+            poll(pollTimeout).apply { commitAsync() }.map { MqTester.Message(it.value()) }
+        }
 
     companion object : KLogging() {
         private const val POLL_MILLIS: Long = 1500
         private const val FETCH_CONSUMER_GROUP_OFFSETS_TIMEOUT: Long = 10
+
+        @JvmField
+        val DEFAULT_CONSUMER_CONFIG: Map<String, String?> = mapOf(
+            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java.name,
+            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java.name,
+            ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to "false",
+            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest",
+        )
     }
 }

exam-mq-rabbit/src/main/java/io/github/adven27/concordion/extensions/exam/mq/rabbit/RabbitTester.kt

@@ -72,13 +72,13 @@ open class RabbitTester @JvmOverloads constructor(
         return result
     }
 
-    override fun send(message: String, headers: Map<String, String>) {
+    override fun send(message: MqTester.Message, params: Map<String, String>) {
         logger.info("Publishing message to {} routing key {} : {}", sendConfig.exchange, sendConfig.routingKey, message)
         ch.basicPublish(
             sendConfig.exchange,
             sendConfig.routingKey,
             MINIMAL_BASIC,
-            sendConfig.sendConverter.apply(MqTester.Message(message, headers))
+            sendConfig.sendConverter.apply(message)
         )
     }
 

exam-mq-redis/src/main/java/io/github/adven27/concordion/extensions/exam/mq/redis/RedisTester.kt

@@ -9,8 +9,8 @@ open class RedisTester(private val host: String, private val port: Int) : MqTest
         jedis = Jedis(host, port)
     }
 
-    override fun send(message: String, headers: Map<String, String>) {
-        val kv = message.split("=").toTypedArray()
+    override fun send(message: MqTester.Message, params: Map<String, String>) {
+        val kv = message.body.split("=").toTypedArray()
         jedis[kv[0].trim { it <= ' ' }] = kv[1].trim { it <= ' ' }
     }