如何向已生成的Protocol Buffers v3消息注入新时间戳字段

问题背景 #

手上有一批遗留的Protocol Buffers (v3)生成消息，其.proto文件无法修改，既用于向下游传输数据，也作为应用内的数据结构。需要在接收消息时记录接收时间戳，但消息及其子消息均无存储时间戳的字段。

核心需求：

• 接收消息时注入新的timestamp字段（支持long或Protobuf Timestamp类型）
• 消息需在应用内传递或继续向下游传输
• 不能定义大量“影子（子）消息”，也不能用包装类（包装类无法解决下游传输兼容性问题）

补充说明 #

应用处于数据处理管道中间环节，接收上游定义的Protobuf消息，处理后传给下游。上游的.proto schema不可修改，下游也依赖原schema实现业务逻辑，因此不能将原消息包装在新消息中。希望通过Protobuf的反射和动态生成类，在运行时解析原消息schema，添加timestamp字段，创建新消息类型，复制原数据并添加时间戳后传递，下游可自行决定是否使用该字段。

Protobuf文档相关说明：

更新消息类型
如果现有消息类型不再满足需求——比如你想给消息格式加一个额外字段——但仍想使用旧格式生成的代码，别担心！使用二进制线格式时，更新消息类型而不破坏现有代码非常简单。

现有代码框架 #

public static <T extends Message> T injectTimestamp(Instant timestamp, T message) {
   // 1. Determine highest/free index number from message
   // 2. Extract message descriptor (builder?)
   // 3. Create new timestamp field using FieldDescriptor builder (?)
   // 4. Add new timestamp field to extracted message descriptor
   // 5. Build new message (type) (with timestamp) from old message using new descriptor
   // 6. Set new timestamp field with timestamp
   // 7. Build new message and return (yay!)
}

完整实现方案 #

核心思路是利用Protobuf的DynamicMessage和反射API，动态修改消息的Descriptor，生成包含新字段的动态消息，同时保证原数据完整迁移。

实现代码 #

import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.google.protobuf.Message;
import com.google.protobuf.Timestamp;
import java.time.Instant;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class ProtobufTimestampInjector {

    public static Message injectTimestamp(Instant timestamp, Message originalMessage) {
        // 步骤1：找到可用的字段编号
        int newFieldNumber = findAvailableFieldNumber(originalMessage.getDescriptorForType());

        // 步骤2：创建timestamp字段的FieldDescriptor
        Descriptors.FieldDescriptor timestampField = createTimestampFieldDescriptor(newFieldNumber);

        // 步骤3：构造新的MessageDescriptor
        Descriptors.Descriptor newMessageDescriptor = createNewMessageDescriptor(originalMessage.getDescriptorForType(), timestampField);

        // 步骤4：将原消息数据复制到新的DynamicMessage.Builder
        DynamicMessage.Builder newMessageBuilder = DynamicMessage.newBuilder(newMessageDescriptor);
        copyOriginalFields(originalMessage, newMessageBuilder);

        // 步骤5：设置timestamp字段值
        Timestamp protoTimestamp = Timestamp.newBuilder()
                .setSeconds(timestamp.getEpochSecond())
                .setNanos(timestamp.getNano())
                .build();
        newMessageBuilder.setField(timestampField, protoTimestamp);

        // 步骤6：构建并返回新消息
        return newMessageBuilder.build();
    }

    private static int findAvailableFieldNumber(Descriptors.Descriptor originalDescriptor) {
        // 取出所有已用字段编号，取最大值+1作为新字段编号
        return originalDescriptor.getFields().stream()
                .mapToInt(Descriptors.FieldDescriptor::getNumber)
                .max()
                .orElse(0) + 1;
    }

    private static Descriptors.FieldDescriptor createTimestampFieldDescriptor(int fieldNumber) {
        return Descriptors.FieldDescriptor.newBuilder()
                .setName("timestamp")
                .setNumber(fieldNumber)
                .setType(Descriptors.FieldDescriptor.Type.TYPE_MESSAGE)
                .setMessageType(Timestamp.getDescriptor())
                .setLabel(Descriptors.FieldDescriptor.Label.LABEL_OPTIONAL)
                .build();
    }

    private static Descriptors.Descriptor createNewMessageDescriptor(Descriptors.Descriptor originalDescriptor, Descriptors.FieldDescriptor newField) {
        // 合并原字段与新字段
        List<Descriptors.FieldDescriptor> newFields = Stream.concat(
                originalDescriptor.getFields().stream(),
                List.of(newField).stream()
        ).collect(Collectors.toList());

        // 构造新的消息描述符
        return Descriptors.Descriptor.newBuilder()
                .setName(originalDescriptor.getName())
                .setFullName(originalDescriptor.getFullName())
                .addAllFields(newFields)
                .build();
    }

    private static void copyOriginalFields(Message originalMessage, DynamicMessage.Builder newMessageBuilder) {
        // 遍历原消息所有字段，复制值到新Builder
        for (Descriptors.FieldDescriptor originalField : originalMessage.getDescriptorForType().getFields()) {
            if (originalMessage.hasField(originalField)) {
                Descriptors.FieldDescriptor newField = newMessageBuilder.getDescriptorForType()
                        .findFieldByName(originalField.getName());
                newMessageBuilder.setField(newField, originalMessage.getField(originalField));
            }
        }
    }
}

关键步骤解析 #

1. 寻找可用字段编号：遍历原消息的所有字段编号，取最大值加1，确保不会与已有字段冲突。
2. 创建字段描述符：定义名为timestamp的可选消息类型字段，关联Protobuf内置的Timestamp类型；如果需要long类型时间戳，可将类型改为TYPE_INT64，直接传入timestamp.toEpochMilli()。
3. 构造新消息描述符：复制原消息的所有字段描述，添加新的timestamp字段，生成新的Descriptor。
4. 复制原数据：通过反射遍历原消息的每个字段，将值复制到新的DynamicMessage.Builder中。
5. 注入时间戳：将Java Instant转换为Protobuf Timestamp，设置到新字段。
6. 生成动态消息：构建DynamicMessage并返回，该消息可直接序列化为二进制流向下游传输，下游用原schema解析时会自动忽略新增字段，完全兼容。

扩展说明 #

如果需要给嵌套的子消息也注入时间戳，可在copyOriginalFields方法中递归处理消息类型的字段，对每个子消息调用injectTimestamp方法。

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