如何在Watermill Kafka库中将Unmarshaler替换为Avro实现反序列化

1. 报错根因说明 #

你遇到的编译错误和反序列化报错分别对应两个核心问题：

• 直接赋值avro.API{}给Unmarshaler字段编译失败：Watermill Kafka的Unmarshaler要求实现对应接口，hamba avro的API原生没有实现该接口，需要自定义封装。
• 消费Java生产的消息报字符串长度小于0：90%以上的跨语言Avro Kafka场景出现该报错，是因为Java侧使用了Confluent官方Avro序列化器，生成的消息前会带5字节固定前缀（1字节魔数0x0 + 4字节Schema ID），直接将完整消息扔给Avro反序列化时，前缀会被当成业务数据解析，读取字符串长度时自然会得到非法负数。你本地Golang生产消费正常，是因为你序列化时没有加这个前缀，格式完全对齐。

2. 具体实现方案 #

步骤1：定义匹配Schema的Golang结构体和预解析Schema #

import (
    "github.com/hamba/avro/v2"
    "github.com/ThreeDotsLabs/watermill-kafka/v2/pkg/kafka"
    "github.com/ThreeDotsLabs/watermill/message"
    "github.com/Shopify/sarama"
    "encoding/json"
)

// 匹配Avro Schema的结构体，注意标签和Schema字段名完全一致
type Avrodata struct {
    ID           int      `avro:"id"`
    TheName      string   `avro:"theName"`
    Dependencies []string `avro:"dependencies"`
}

// 预解析Avro Schema，全局初始化一次即可
var avroSchema = avro.MustParse(`{
    "namespace": "data.avro",
    "doc":"Docstring.",
    "name": "Avrodata",
    "type": "record",
    "fields": [
        {"name": "id",       "type": "int"},
        {"name": "theName",      "type": "string"},
        {"name": "dependencies",  "type": {"type": "array", "items": "string"}}
    ]
}`)

步骤2：自定义实现Watermill的Unmarshaler接口 #

type AvroUnmarshaler struct {
    Schema avro.Schema
    // 若使用Confluent Schema Registry可在此添加Registry客户端字段
}

func (a AvroUnmarshaler) Unmarshal(saramaMsg *sarama.ConsumerMessage, msg *message.Message) error {
    var data Avrodata
    // 核心：如果Java侧用了Confluent序列化器带前缀，就把saramaMsg.Value改成saramaMsg.Value[5:]
    payload := saramaMsg.Value
    // payload = saramaMsg.Value[5:] // 确认有前缀时打开这行注释即可
    err := avro.Unmarshal(a.Schema, payload, &data)
    if err != nil {
        return err
    }
    // 可根据需求将解析后的数据放入Payload或者Message的Context中
    // 示例中直接序列化后放入Payload，也可自定义传递方式
    msg.Metadata.Set("id", string(rune(data.ID)))
    msg.Metadata.Set("theName", data.TheName)
    msg.Payload, err = json.Marshal(data)
    return err
}

步骤3：替换Subscriber配置中的Unmarshaler #

subscriberConfig := kafka.SubscriberConfig{
    Brokers:               []string{"localhost:9092"},
    // 替换为自定义的AvroUnmarshaler
    Unmarshaler:           AvroUnmarshaler{Schema: avroSchema},
    OverwriteSaramaConfig: saramaSubscriberConfig,
    ConsumerGroup:         "test_consumer_group",
}

3. 额外注意事项 #

• 若Schema有更新，需要同步更新Golang结构体和预解析的Schema内容
• 确认Java侧是否使用了Schema Registry，若使用则必须跳过前5字节前缀，否则会持续出现解析错误
• 结构体的Avro标签必须和Schema字段名大小写完全匹配，否则会出现字段映射为空的问题

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