如何用Kotlin消费单Kafka Topic中不同Avro命名空间的对象？

你遇到的核心问题是：KafkaAvroDeserializer默认返回的对象类型和你预期的Dog/Cat类不匹配，导致is Dog/is Cat的类型判断始终为false，最终过滤出空列表。以下是具体解决思路和实现方案：

为什么类型判断失效？ #

KafkaAvroDeserializer有两种工作模式：

1. 默认模式：返回GenericRecord对象（不管你是否有对应的实体类），此时record.value() is Dog必然不成立，因为实际类型是GenericRecord。
2. SpecificRecord模式：需配置specific.avro.reader=true，但只有当你的Dog/Cat类是严格按照Topic中Avro Schema生成的（命名空间、字段、类型完全匹配），才会自动映射成对应类；否则仍返回GenericRecord。

解决方案一：基于SpecificRecord自动映射 #

步骤： #

1. 用Avro工具（如Maven/Gradle的Avro插件），根据Topic中的Avro Schema生成Dog/Cat的实体类，确保类的命名空间、字段、类型完全匹配Schema定义。
2. 在消费者配置中添加参数：

   specific.avro.reader=true
   

3. 修改消费代码的泛型类型（如果使用框架，需确保框架传递的是正确的类型）：

   @Incoming("some-ingest")
   fun consumeMyTopic(records: ConsumerRecords<String, SpecificRecord>) {
       val recordsDogs = records.filter { it.value() is Dog }
           .map { it as ConsumerRecord<String, Dog> }
           .map { toDog(it, DogMapper::convertToEntity) }
       
       val recordsCats = records.filter { it.value() is Cat }
           .map { it as ConsumerRecord<String, Cat> }
           .map { toProgramme(it, CatMapper::convertToEntity) }
       
       dogService.saveDogs(recordsDogs)
       catService.saveCats(recordsCats)
   }
   

解决方案二：直接处理GenericRecord（无需生成实体类） #

如果无法生成完全匹配的SpecificRecord类，可通过GenericRecord的Schema信息判断类型，手动转换为你的业务实体：

import org.apache.avro.generic.GenericRecord

@Incoming("some-ingest")
fun consumeMyTopic(records: ConsumerRecords<String, GenericRecord>) {
    if (logger.isDebugEnabled) {
        logger.debugf("some-ingest: consuming %d record(s)", records.count())
    }

    // 替换为Topic中Dog/Cat Schema的完整命名（如com.example.avro.Dog）
    val dogSchemaFullName = "your.dog.schema.full.name"
    val catSchemaFullName = "your.cat.schema.full.name"

    val recordsDogs = records.filter { 
        it.value().schema.fullName == dogSchemaFullName 
    }.map { record ->
        // 手动将GenericRecord字段映射到Dog实体
        val dogEntity = Dog(
            id = record.get("id") as String,
            name = record.get("name") as String
            // 其他字段按需映射
        )
        toDog(record, DogMapper::convertToEntity) // 复用原有转换逻辑
    }

    val recordsCats = records.filter { 
        it.value().schema.fullName == catSchemaFullName 
    }.map { record ->
        val catEntity = Cat(
            id = record.get("id") as String,
            color = record.get("color") as String
            // 其他字段按需映射
        )
        toProgramme(record, CatMapper::convertToEntity)
    }

    dogService.saveDogs(recordsDogs)
    catService.saveCats(recordsCats)
}

解决方案三：自定义反序列化器（灵活映射） #

如果需要更灵活的类型映射，可继承KafkaAvroDeserializer，根据Schema信息自动转换为业务实体：

1. 自定义反序列化器 #

import io.confluent.kafka.serializers.KafkaAvroDeserializer
import org.apache.avro.generic.GenericRecord

class MultiAvroDeserializer : KafkaAvroDeserializer() {
    override fun deserialize(topic: String?, data: ByteArray?): Any {
        val genericRecord = super.deserialize(topic, data) as GenericRecord
        return when (genericRecord.schema.fullName) {
            "your.dog.schema.full.name" -> convertToDog(genericRecord)
            "your.cat.schema.full.name" -> convertToCat(genericRecord)
            else -> genericRecord // 或抛出异常，根据业务需求处理
        }
    }

    private fun convertToDog(record: GenericRecord): Dog {
        return Dog(
            id = record.get("id") as String,
            name = record.get("name") as String
            // 其他字段映射
        )
    }

    private fun convertToCat(record: GenericRecord): Cat {
        return Cat(
            id = record.get("id") as String,
            color = record.get("color") as String
            // 其他字段映射
        )
    }
}

2. 配置消费者使用自定义反序列化器 #

value.deserializer=your.package.MultiAvroDeserializer

3. 消费代码恢复原有逻辑 #

此时record.value()会自动是Dog/Cat类型，原有过滤和转换代码即可正常工作。

排查要点 #

• 确认消费者配置的schema.registry.url正确，能正常拉取Topic对应的Schema。
• 可先打印record.value().javaClass.name，查看实际返回的对象类型，再针对性调整方案。
• 如果使用SpecificRecord模式，确保生成的实体类和Schema的命名空间、字段完全一致。

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