C#消费Kafka消息时Protobuf反序列化报错：缺失对应起始组标签

报错信息 #

Merge an unknown field of end-group tag, indicating that the corresponding start-group was missing.

问题描述 #

将Kafka消息保存为二进制文件时Python可正常处理，但C#中反序列化报错。以下是C#消费代码：

using System;
using Confluent.Kafka;
using Google.Protobuf;
using RequestCreatedEvent;

class Program {
    static void Main(string[] args) {
    var config = new ConsumerConfig {
        GroupId = "testcdm",
            BootstrapServers = "localhost:9092",
            AutoOffsetReset = AutoOffsetReset.Latest
    };

    using(var consumer = new ConsumerBuilder < Ignore, string > (config).Build()) {
        consumer.Subscribe("dest-topic");

        Console.WriteLine("Consumer started. Waiting for messages...");

        try {
            while (true) {
                var consumeResult = consumer.Consume(TimeSpan.FromSeconds(1));
                if (consumeResult != null) {
                    Console.WriteLine($"Received message: {consumeResult.Message.Value}");

                    // Decoding Base64 data
                    byte[] decodedData = Convert.FromBase64String(consumeResult.Message.Value);

                    try {
                        // Deserializing the Protobuf message
                        var requestCreated = RequestCreated.Parser.ParseFrom(decodedData);

                        // Accessing fields from the Protobuf message
                        Console.WriteLine($"systemId: {requestCreated.Merchant.SystemId}");
                    } catch (Exception ex) {
                        Console.WriteLine($"Protobuf deserialization error: {ex.Message}");
                    }
                }
            }
        } catch (ConsumeException e) {
            Console.WriteLine($"Consume error: {e.Error.Reason}");
        } catch (Exception e) {
            Console.WriteLine($"Unexpected error: {e.Message}");
        } finally {
            consumer.Close();
        }
    }
  }
}

实际输出报错 #

Output : Consumer started. Waiting for messages...

Received message:

ClcKJGY5ZjZkZDg5LTc3MWItNTExMy1hNzViLTE5NjRhYWQxOGFjOCACMisKDAiLho61BhDA84ipAhINRXVyb3BlL1p1cmljaBoMCIuGjrUGEMDziKkCOAFaCURhdGF0cmFuc2IrCgwIq76OtQYQwMPFmgISDUV1cm9wZS9adXJpY2gaDAirvo61BhDAw8WaAmoWCg4zNS4xNTcuMTg5LjIzMBIERWNvbXIUEhIyMTA3MTMxMjQ4MDczNTUzMzh6BgoERWNvbYIBEgoQCgNDSEYQAhkAAAAAAACuQIoBJDoiCiA3TEhYc2Nxd0FBRUFBQUdNN3A0bEpHdmdtZEJRQUZRT5IBRBIPCgt1c2VyX2RhdGFfMRIAEg8KC3VzZXJfZGF0YV8yEgASDwoLdXNlcl9kYXRhXzMSABIPCgt1c2VyX2RhdGFfNBIA

Protobuf deserialization error: Merge an unknown field of end-group tag, indicating that the corresponding start-group was missing.

可能的原因及解决方法 #

1. Protobuf版本或语法不兼容 #

• 原因：Python和C#使用的Protobuf库版本差异大，或者.proto文件语法版本（proto2/proto3）不一致，导致序列化格式存在差异。Python的Protobuf库对格式容错性较高，而C#库更严格。
• 解决：
  • 确保两端使用完全相同的.proto定义文件，统一语法版本（比如都用proto3）。
  • 同步Python和C#的Protobuf库版本，尽量使用最新稳定版。
  • 重新生成C#端的Protobuf代码，使用对应语法版本的生成命令（比如proto3用protoc --csharp_out=. your.proto）。

2. 消息序列化/解析方式不匹配 #

• 原因：Python端可能使用了带长度前缀的序列化方式（比如SerializeDelimitedTo），而C#端直接用ParseFrom解析无长度前缀的二进制流，导致格式错位。
• 解决：
  • 检查Python端的序列化代码，如果是带长度前缀的消息，C#端改用ParseDelimitedFrom解析：

    using (var stream = new MemoryStream(decodedData))
    {
        var requestCreated = RequestCreated.Parser.ParseDelimitedFrom(stream);
        // 后续处理逻辑
    }
    

3. Base64编码格式差异 #

• 原因：Python端可能使用了URL安全的Base64编码（替换+为-、/为_），而C#用标准Base64解码，导致二进制数据错误。
• 解决：
  • 在C#解码前先还原URL安全的Base64字符：

    string base64Str = consumeResult.Message.Value.Replace('-', '+').Replace('_', '/');
    byte[] decodedData = Convert.FromBase64String(base64Str);
    

4. Protobuf代码生成错误 #

• 原因：C#端的Protobuf代码生成不完整，或者启用了错误的编译选项（比如旧版的group特性，而.proto中并未使用）。
• 解决：
  • 清理旧的生成代码，用最新版的protoc工具重新生成C#代码。
  • 检查生成代码是否与.proto文件定义完全对应，比如嵌套消息、字段类型是否一致。

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