Citrus Kafka端点接收AVRO消息转JSON并做JsonPath验证的问题

问题核心 #

使用Citrus+Kotlin+Quarkus测试Kafka AVRO主题时，尝试用JsonPath验证消息内容却报错，本质是Citrus无法直接将AVRO反序列化后的对象识别为JSON格式，导致找不到对应验证器。即使换用StringDeserializer，AVRO字节流转成的字符串并非有效JSON，同样无法触发JSON验证器。

解决方案步骤 #

1. 引入必要依赖 #

添加Jackson AVRO模块，用于将AVRO对象（GenericRecord/AVRO POJO）转换为JSON字符串：

<!-- Quarkus pom.xml 中添加 -->
<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-avro</artifactId>
</dependency>

<!-- 确保Citrus JSON验证器依赖已引入 -->
<dependency>
    <groupId>com.consol.citrus</groupId>
    <artifactId>citrus-json</artifactId>
    <version>${citrus.version}</version>
    <scope>test</scope>
</dependency>

2. 正确配置Kafka消费者属性 #

确保KafkaAvroDeserializer的参数配置完整，能正确反序列化AVRO消息：

val consumerPropertiesMap = mapOf(
    "bootstrap.servers" to "boostrap.server-1,boostrap.server-2,boostrap.server-3",
    "schema.registry.url" to "你的Schema Registry地址",
    "basic.auth.credentials.source" to "USER_INFO",
    "basic.auth.user.info" to "用户名:密码",
    "specific.avro.reader" to "false" // 用GenericRecord则设为false，用自定义AVRO POJO设为true
)

3. 添加AVRO转JSON的消息转换器 #

自定义Citrus MessageConverter，自动将AVRO对象转换为JSON字符串，让Citrus能识别并使用JSON验证器：

import com.consol.citrus.message.Message
import com.consol.citrus.message.MessageBuilder
import com.consol.citrus.message.MessageConverter
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.dataformat.avro.AvroModule
import org.apache.avro.generic.GenericRecord

val avroToJsonConverter = object : MessageConverter {
    private val objectMapper = ObjectMapper().registerModule(AvroModule())

    override fun canConvert(message: Message<*>?, targetType: Class<*>?): Boolean {
        // 仅处理AVRO GenericRecord转JSON字符串的场景
        return targetType == String::class.java && message?.payload is GenericRecord
    }

    override fun <T> convert(message: Message<*>?, targetType: Class<T>?): Message<T> {
        val jsonPayload = objectMapper.writeValueAsString(message?.payload)
        return MessageBuilder.withPayload(jsonPayload as T).build()
    }
}

// 初始化KafkaEndpoint时绑定转换器
val eventsTopicEndpoint = KafkaEndpointBuilder()
    .server("boostrap.server-1,boostrap.server-2,boostrap.server-3")
    .topic("events.topic.with.avro.values")
    .consumerProperties(consumerPropertiesMap)
    .valueDeserializer(KafkaAvroDeserializer::class.java)
    .messageConverter(avroToJsonConverter) // 绑定自定义转换器
    .build()

4. 修正验证代码 #

现在可以正常使用JsonPath验证JSON格式的消息内容：

testCaseRunner.then(receive()
    .endpoint(eventsTopicEndpoint)
    .message()
    .type(MessageType.JSON)
    .validate(jsonPath()
        .expression("$.eventId", contains(file.name))))

替代方案：手动转换消息体 #

如果不想全局配置转换器，也可以在接收消息时手动转换：

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.dataformat.avro.AvroModule
import org.apache.avro.generic.GenericRecord

testCaseRunner.then(receive()
    .endpoint(eventsTopicEndpoint)
    .transformer { message ->
        val avroRecord = message.payload as GenericRecord
        val jsonString = ObjectMapper().registerModule(AvroModule())
            .writeValueAsString(avroRecord)
        MessageBuilder.withPayload(jsonString).build()
    }
    .message()
    .type(MessageType.JSON)
    .validate(jsonPath()
        .expression("$.eventId", contains(file.name))))

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