.NET 7读取Kafka ProtoBuf墓碑消息反序列化报错求助

问题原因 #

你遇到的报错是因为Confluent官方的ProtobufDeserializer默认只处理符合Protobuf编码规范的有效数据，而Kafka墓碑消息的空负载（0字节）不符合它的预期——该Deserializer默认期望数据带有至少6字节的framing结构（Confluent Protobuf序列化的约定），所以空数据直接触发异常，即便Consumer层传递了空Span，Deserializer内部也没有处理这种墓碑消息的场景。

解决思路：自定义包装Deserializer处理空负载 #

你需要实现一个包装类，在调用原ProtobufDeserializer之前先判断是否是墓碑消息（空值/空数据），如果是则直接返回null，否则再交给原Deserializer处理。

自定义Deserializer代码 #

public class TombstoneProtobufDeserializer<T> : IAsyncDeserializer<T> where T : class, IMessage<T>, new()
{
    private readonly ProtobufDeserializer<T> _innerDeserializer;

    public TombstoneProtobufDeserializer()
    {
        _innerDeserializer = new ProtobufDeserializer<T>();
    }

    public async Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool isNull, SerializationContext context)
    {
        // 匹配墓碑消息：isNull为true 或 数据长度为0时返回null
        if (isNull || data.Length == 0)
        {
            return null;
        }
        // 非空数据交给原ProtobufDeserializer处理
        return await _innerDeserializer.DeserializeAsync(data, isNull, context);
    }
}

修改Consumer初始化代码 #

把原来的ProtobufDeserializer替换成自定义的包装类：

var consumer = new ConsumerBuilder<string, Profile>(consumerConfig)
    .SetValueDeserializer(new TombstoneProtobufDeserializer<Profile>().AsSyncOverAsync())
    .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
    .Build();

额外验证点 #

• 确认Kafka中的墓碑消息确实是**键存在、值为空（0字节）**的记录，避免是生产者发送时的其他错误导致的空数据。
• 确保你的Profile类是Protobuf工具生成的，且实现了IMessage<Profile>接口（这是Protobuf生成类的默认实现），否则自定义Deserializer的泛型约束会不匹配。

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