Microsoft Avro序列化器篡改传入Schema问题求助

一、现有库的修复/规避方法 #

1. 手动修正输出文件的Schema #

Avro文件开头包含Schema的JSON定义，你可以在序列化完成后手动替换被篡改的部分：

• 跳过文件开头4字节的magic标识（"Obj\x01"），读取变长整数获取Schema的长度，再读取对应长度的JSON字符串。
• 将JSON中被替换的"type": "long"替换回原始的{"type": "long", "logicalType": "timestamp-micros"}，注意匹配字段上下文，避免误替换其他无关的long类型定义。
• 将修改后的Schema写回文件对应位置，覆盖原内容。

简化示例代码：

using (var fs = new FileStream("output.avro", FileMode.Open, FileAccess.ReadWrite))
{
    // 跳过magic bytes
    fs.Seek(4, SeekOrigin.Begin);
    // 读取Schema长度（Avro变长整数编码）
    int schemaLength = ReadVarInt(fs);
    // 读取原始Schema JSON
    byte[] schemaBytes = new byte[schemaLength];
    fs.Read(schemaBytes, 0, schemaLength);
    string schemaJson = Encoding.UTF8.GetString(schemaBytes);
    // 替换篡改的Schema片段
    string correctedSchema = schemaJson.Replace(
        "\"timestamp\": {\"type\": \"long\"}", 
        "\"timestamp\": {\"type\": \"long\", \"logicalType\": \"timestamp-micros\"}"
    );
    // 写回修正后的Schema
    fs.Seek(4 + GetVarIntLength(schemaLength), SeekOrigin.Begin);
    byte[] correctedBytes = Encoding.UTF8.GetBytes(correctedSchema);
    fs.Write(correctedBytes, 0, correctedBytes.Length);
}

（需自行实现ReadVarInt和GetVarIntLength方法，遵循Avro变长整数编码规则）

2. 反射修改库内部逻辑 #

反编译Microsoft.Avro.Core的代码，找到负责解析/生成Schema的类（如Schema或SchemaBuilder），通过反射修改其处理logicalType的逻辑：

• 定位到剥离logicalType、提取内部type字段的代码段。
• 强制修改逻辑，当遇到包含logicalType的类型对象时，保留完整的Schema结构而非仅提取内部type。

此方法需要熟悉.NET反射机制，适合有一定底层开发经验的开发者。

二、替代库推荐（支持动态类型+维护活跃） #

1. Apache Avro .NET #

官方Avro的.NET实现，原生支持动态类型序列化：

• 使用GenericRecord处理编译时未知的类型，根据Schema动态构建记录。
• 完全兼容timestamp-micros等logicalType，不会篡改Schema定义。
• 示例代码：

var schema = Schema.Parse(@"{""type"": ""record"", ""name"": ""Test"", ""fields"": [{""name"": ""timestamp"", ""type"": {""type"": ""long"", ""logicalType"": ""timestamp-micros""}}]}");
var record = new GenericRecord(schema);
record.Add("timestamp", DateTime.UtcNow);

using (var fs = new FileStream("output.avro", FileMode.Create))
{
    using (var writer = AvroWriter.CreateGeneric(schema, fs))
    using (var streamWriter = new SequentialWriter(writer, 4096))
    {
        streamWriter.Write(record);
    }
}

2. Confluent.Kafka.Avro #

基于官方库扩展，除基础Avro序列化外，支持Schema Registry，同样适配动态类型场景：

• 适合与Kafka结合的业务场景，也可单独用于文件序列化。
• 对logicalType的支持完善，Schema输出完全符合规范。

3. AvroConvert #

轻量级Avro序列化库，支持无预定义类型的动态序列化：

• API简洁，直接传入对象和Schema即可完成序列化/反序列化。
• 维护活跃，兼容最新Avro规范及各类logicalType。

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