如何在ReadyAPI中验证Kafka特定Key消息仅到达一次？

一、Script Assertion问题逐个解决 #

1. 无法创建计数文件 #

指定绝对路径并确保目录可写，避免相对路径权限问题。可通过System.getProperty("user.dir")获取ReadyAPI项目根目录，提前创建计数文件存放目录，同时捕获IO异常。

2. 无法打印ReadyAPI日志 #

直接使用内置log对象，调用log.info()/log.error()方法即可输出到ReadyAPI日志面板。

3. 无法访问项目属性 #

通过context对象层级获取项目属性：

def projectProperty = context.testCase.testSuite.project.getPropertyValue("你的属性名")

确保属性已在ReadyAPI项目的「Custom Properties」中定义。

4. 获取Kafka消息Key与解析JSON #

• 获取Kafka消息Key：如果是Kafka Consumer步骤返回的消息，直接调用message.getKey()；部分版本可通过消息头获取：message.getHeaders().getFirst("kafka_key")
• 解析JSON：使用Groovy内置JsonSlurper，无需额外引入类库，直接解析消息内容。

二、完整Script Assertion示例 #

import groovy.json.JsonSlurper
import java.io.File
import java.io.FileWriter
import java.io.BufferedReader

// 1. 获取当前Kafka消息的Key
def messageKey = message.getKey()?.toString()
if (!messageKey) {
    assertionStatus = false
    assertionFailureMessage = "未获取到Kafka消息Key"
    log.error(assertionFailureMessage)
    return
}

// 2. 读取项目属性（示例：获取计数文件存储路径）
def countFilePath = context.testCase.testSuite.project.getPropertyValue("KafkaMessageCountPath")
if (!countFilePath) {
    // 若未配置属性，使用项目目录默认路径
    countFilePath = System.getProperty("user.dir") + "/kafka_message_counts.txt"
}
def countFile = new File(countFilePath)

// 3. 初始化或读取计数文件
def keyCountMap = [:]
if (countFile.exists()) {
    def reader = new BufferedReader(new FileReader(countFile))
    def line
    while ((line = reader.readLine()) != null) {
        def parts = line.split("=")
        if (parts.length == 2) {
            keyCountMap[parts[0].trim()] = parts[1].trim().toInteger()
        }
    }
    reader.close()
}

// 4. 更新当前Key的计数
def currentCount = keyCountMap.getOrDefault(messageKey, 0) + 1
keyCountMap.put(messageKey, currentCount)

// 5. 写入计数文件到磁盘
def writer = new FileWriter(countFile)
keyCountMap.each { key, count ->
    writer.write("${key}=${count}\n")
}
writer.close()

// 6. 解析JSON消息内容（可选，验证消息内容正确性）
def jsonSlurper = new JsonSlurper()
try {
    def messageContent = jsonSlurper.parseText(message.getPayload())
    log.info("解析的消息内容：${messageContent}")
    // 可添加额外的内容验证逻辑，比如检查特定字段
} catch (Exception e) {
    log.error("JSON解析失败：${e.message}")
}

// 7. 核心断言：验证当前Key的消息仅发送一次
if (currentCount != 1) {
    assertionStatus = false
    assertionFailureMessage = "Kafka消息Key【${messageKey}】发送次数异常，当前次数：${currentCount}"
    log.error(assertionFailureMessage)
} else {
    assertionStatus = true
    log.info("Kafka消息Key【${messageKey}】发送次数验证通过，当前次数：${currentCount}")
}

三、使用说明 #

1. 在ReadyAPI的Kafka Consumer步骤中添加「Script Assertion」，粘贴上述代码
2. 可选：在项目属性中添加KafkaMessageCountPath，指定计数文件的存储路径
3. 若需要重置计数，直接删除计数文件即可
4. 结合TestComplete调用时，确保ReadyAPI项目路径正确，计数文件有读写权限

内容的提问来源于stack exchange，提问作者primehunter