如何无需Schema Registry实现Avro Schema兼容性单元测试？

针对你不想依赖Docker运行Schema Registry来测试Avro Schema兼容性的需求，有两种直接可行的方案：

方案一：实现内存版ISchemaRegistry #

你可以自己编写一个极简的内存版ISchemaRegistry实现，满足序列化/反序列化过程中对Schema注册、查找的需求，完全不需要外部服务。

内存版Registry实现示例 #

public class InMemorySchemaRegistry : ISchemaRegistry
{
    private readonly Dictionary<int, string> _schemaById = new();
    private readonly Dictionary<string, int> _schemaIdByContent = new();
    private int _nextId = 1;

    public Task<int> RegisterSchemaAsync(string subject, string schema, CancellationToken cancellationToken = default)
    {
        if (_schemaIdByContent.TryGetValue(schema, out var id))
        {
            return Task.FromResult(id);
        }
        
        id = _nextId++;
        _schemaById[id] = schema;
        _schemaIdByContent[schema] = id;
        return Task.FromResult(id);
    }

    public Task<string> GetSchemaAsync(int id, CancellationToken cancellationToken = default)
    {
        if (_schemaById.TryGetValue(id, out var schema))
        {
            return Task.FromResult(schema);
        }
        throw new SchemaRegistryException($"Schema with id {id} not found");
    }

    public Task<SchemaVersion> GetLatestSchemaAsync(string subject, CancellationToken cancellationToken = default)
    {
        // 测试场景按需返回对应版本Schema，比如返回旧版本
        var latestSchema = _schemaById.Values.Last();
        return Task.FromResult(new SchemaVersion(latestSchema, _nextId - 1, subject));
    }

    // 其他未用到的接口方法可按需实现或抛出NotImplementedException
}

测试代码改造 #

预先将新旧Schema注册到内存Registry中，再执行兼容性测试：

// 从avsc文件读取或通过生成代码获取新旧Schema内容
string oldSchema = File.ReadAllText("MyModelOld.avsc");
string newSchema = File.ReadAllText("MyModel.avsc");

var schemaRegistry = new InMemorySchemaRegistry();
await schemaRegistry.RegisterSchemaAsync("my-model-subject", oldSchema);
await schemaRegistry.RegisterSchemaAsync("my-model-subject", newSchema);

var deserializerOld = new AvroDeserializer<MyModelOld>(schemaRegistry);
var serializerNew = new AvroSerializer<MyModel>(schemaRegistry);

var myModel = new MyModel() { /* 初始化包含新增属性的实例 */ };
var myModelBytes = await serializerNew.SerializeAsync(myModel, new());

var myModelOld = await deserializerOld.DeserializeAsync(myModelBytes, false, new());
// 验证旧模型的原有属性是否正确赋值

方案二：绕开Schema Registry，直接使用Schema对象 #

如果测试不需要模拟Schema Registry的版本管理逻辑，可以直接基于Avro Schema对象创建序列化器/反序列化器，完全跳过Registry依赖。

测试代码示例 #

// 解析新旧Schema文件
var oldSchema = Avro.Schema.Parse(File.ReadAllText("MyModelOld.avsc"));
var newSchema = Avro.Schema.Parse(File.ReadAllText("MyModel.avsc"));

// 正向兼容性测试：新模型序列化，旧模型反序列化
var serializerNew = new AvroSerializer<MyModel>(newSchema);
var deserializerOld = new AvroDeserializer<MyModelOld>(oldSchema);

var myModel = new MyModel() { /* 初始化属性 */ };
var myModelBytes = await serializerNew.SerializeAsync(myModel, new());
var myModelOld = await deserializerOld.DeserializeAsync(myModelBytes, false, new());
// 验证旧模型属性正确性

// 反向兼容性测试：旧模型序列化，新模型反序列化（验证新增属性默认值）
var serializerOld = new AvroSerializer<MyModelOld>(oldSchema);
var deserializerNew = new AvroDeserializer<MyModel>(newSchema);

var myModelOldInstance = new MyModelOld() { /* 初始化旧属性 */ };
var oldModelBytes = await serializerOld.SerializeAsync(myModelOldInstance, new());
var myModelNew = await deserializerNew.DeserializeAsync(oldModelBytes, false, new());
// 验证新增属性是否使用Schema定义的默认值

关于Mock ISchemaRegistry的问题 #

你之前Mock失败的原因是，Avro序列化/反序列化内部会调用Registry的多个核心方法（Schema注册、ID查找、版本获取等），Mock需要完整实现这些逻辑才能正常工作，不如直接编写极简内存实现或绕开Registry更高效。

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