如何解决自托管Kafka的AWS EventBridge Pipe Protobuf事件反序列化问题

问题背景 #

我配置了EventBridge Pipe从自托管Kafka消费事件并转发至Step Function，Kafka Topic中存储的是Protobuf格式消息。Pipe传递的事件结构如下：

[
  {
    "topic": "<REDUCTED>",
    "partition": 0,
    "offset": 226,
    "timestamp": 1694180898000,
    "timestampType": "CREATE_TIME",
    "key": "ABCDVE1BTi0yMDIzLTA5LTA3",
    "value": "AAAAFVwAChJ .......... pZGVudCBub24gAoABAQ==",
    "headers": [],
    "bootstrapServers": "smk://<REDUCTED>",
    "eventSource": "SelfManagedKafka",
    "eventSourceKey": "<REDUCTED>"
  }
]

我在Pipe中添加了富化工序，调用基于Quarkus构建的Lambda函数将Protobuf转为JSON，以便Step Function进一步处理，但多次尝试解析value字段的Protobuf模型均失败：

尝试的解析方法及结果 #

• 方法1：使用ByteBuffer的parseFrom

ByteBuffer wrap = ByteBuffer.wrap(value.getBytes(StandardCharsets.UTF_8));
return MyEventProto.MyEvent.parseFrom(wrap);

错误信息：

com.google.protobuf.InvalidProtocolBufferException: 
  Protocol message end-group tag did not match expected tag.

• 方法2：使用字节流的parseDelimitedFrom

ByteArrayInputStream inputStream = new ByteArrayInputStream(value.getBytes(StandardCharsets.UTF_8));
return MyEventProto.MyEvent.parseDelimitedFrom(inputStream);

错误信息：

com.google.protobuf.InvalidProtocolBufferException: 
  While parsing a protocol message, the input ended unexpectedly in the middle of a field.  
  This could mean either that the input has been truncated or that an embedded message misreported 
  its own length.

• 方法3：Base64解码后使用parseDelimitedFrom/parseFrom

byte[] decode = Base64.getDecoder().decode(value.getBytes(StandardCharsets.UTF_8));
return MyEventProto.MyEvent.parseDelimitedFrom(new ByteArrayInputStream(decode));
// 或 return MyEventProto.MyEvent.parseFrom(new ByteArrayInputStream(decode));

结果：无报错，但返回空对象，与MyEventProto.MyEvent.getDefaultInstance()返回结果一致

• 方法4：Base64解码后用parseFrom

byte[] decode = Base64.getDecoder().decode(value.getBytes(StandardCharsets.UTF_8));
return MyEventProto.MyEvent.parseFrom(decode);

错误信息：

com.google.protobuf.InvalidProtocolBufferException: 
  Protocol message contained an invalid tag (zero).

补充尝试（编辑1） #

查阅文档得知AWS传递的是Base64编码消息，调整代码如下：

byte[] bytes = Base64.getDecoder().decode(value.getBytes());
ByteBuffer wrap = ByteBuffer.wrap(bytes);
return MyEventProto.MyEvent.parseFrom(wrap);

测试发现，手动移除开头的AAAAFVwA和末尾的==后可成功解析，但不清楚开头额外字节的来源。

正确解析方案 #

问题根源在于AWS传递的value字段解码后包含Kafka的消息封装前缀，而非纯Protobuf payload。

标准Kafka消息的二进制结构默认包含7字节前缀：

1. 魔术字节（1字节，标识Kafka消息版本）
2. 消息属性（2字节，如压缩类型）
3. 消息体长度（4字节）

你需要先跳过这些前缀，再解析Protobuf内容：

import java.util.Arrays;
import java.util.Base64;

// 解码Base64字符串
byte[] decodedBytes = Base64.getDecoder().decode(value);

// 跳过Kafka默认7字节前缀
int kafkaPrefixLength = 7;
byte[] protobufPayload = Arrays.copyOfRange(decodedBytes, kafkaPrefixLength, decodedBytes.length);

// 解析纯Protobuf数据
return MyEventProto.MyEvent.parseFrom(protobufPayload);

说明 #

• 你手动移除的AAAAFVwA正好对应Base64编码后的7字节Kafka前缀，验证了这个逻辑。
• Base64末尾的==是编码填充符，解码器会自动处理，无需手动移除。
• 若你的Kafka集群使用了自定义消息格式（如自定义头、特殊压缩配置），前缀长度可能变化，需根据实际配置调整跳过的字节数。

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