将JSON转换为Protobuf DynamicMessage时Unix时间戳转换失败

我正在尝试将JSON字符串动态转换为Protobuf的DynamicMessage类型，使用的Kotlin代码如下：

val topic = "sample-topic"
// 使用Confluent Schema Registry存储/获取Schema
val descriptor: Descriptors.Descriptor = getDescriptorFromRegistry(topic)
// 创建DynamicMessage Builder并传入对应的Descriptor
val builder: DynamicMessage.Builder = DynamicMessage.newBuilder(descriptor)
// 将JSON与Descriptor合并
JsonFormat.parser().merge(json, builder) // 此处抛出异常
val message: DynamicMessage = builder.build()
val bytes: ByteArray = message.toByteArray()

对应的JSON字符串：

{
    "id": 1001,
    "firstName": "Bob",
    "lastName": "Test",
    "email": "bob@test.com",
    "dateCreated": 1696273647694046
}

Protobuf Schema定义：

syntax = "proto3";
package io.smyrgeorge.test.proto.domain;

import "google/protobuf/timestamp.proto";

message Customer {
  int32 id = 1;
  string firstName = 2;
  string lastName = 3;
  string email = 4;
  google.protobuf.Timestamp dateCreated = 5;
}

转换时因dateCreated字段抛出异常，该字段是微秒级Unix时间戳1696273647694046，而Protobuf默认解析器仅支持RFC 3339格式的ISO日期字符串，异常信息如下：

Exception in thread "main" com.google.protobuf.InvalidProtocolBufferException: Failed to parse timestamp: 1696273647694046
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeTimestamp(JsonFormat.java:1564)
at com.google.protobuf.util.JsonFormat$ParserImpl.access$2200(JsonFormat.java:1299)
at com.google.protobuf.util.JsonFormat$ParserImpl$3.merge(JsonFormat.java:1394)
at com.google.protobuf.util.JsonFormat$ParserImpl.merge(JsonFormat.java:1455)
at com.google.protobuf.util.JsonFormat$ParserImpl.parseFieldValue(JsonFormat.java:2018)
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeField(JsonFormat.java:1669)
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeMessage(JsonFormat.java:1500)
at com.google.protobuf.util.JsonFormat$ParserImpl.merge(JsonFormat.java:1458)
at com.google.protobuf.util.JsonFormat$ParserImpl.parseFieldValue(JsonFormat.java:2018)
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeOneofField(JsonFormat.java:1708)
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeField(JsonFormat.java:1667)
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeMessage(JsonFormat.java:1500)
at com.google.protobuf.util.JsonFormat$ParserImpl.merge(JsonFormat.java:1458)
at com.google.protobuf.util.JsonFormat$ParserImpl.merge(JsonFormat.java:1340)
at com.google.protobuf.util.JsonFormat$Parser.merge(JsonFormat.java:472)
at io.smyrgeorge.connect.converter.ProtobufConverter.fromConnectData(ProtobufConverter.kt:93)
at io.smyrgeorge.connect.MainKt.main(Main.kt:54)
at io.smyrgeorge.connect.MainKt.main(Main.kt)
Caused by: java.text.ParseException: Failed to parse timestamp: invalid timestamp "1696273647694046"
at com.google.protobuf.util.Timestamps.parse(Timestamps.java:241)
at com.google.protobuf.util.JsonFormat$ParserImpl.mergeTimestamp(JsonFormat.java:1561)
... 17 more

我希望避免预处理JSON字符串，请问是否可以为Timestamp类型传入自定义解析器？

Protobuf官方的JsonFormat.Parser没有直接提供自定义内置类型（如Timestamp）解析逻辑的扩展点，但可以通过先解析JSON为结构化对象，手动处理Timestamp字段，再合并剩余字段的方式实现，无需修改原始JSON字符串：

自定义合并逻辑实现 #

import com.google.gson.JsonParser
import com.google.protobuf.util.JsonFormat
import com.google.protobuf.util.Timestamps
import com.google.protobuf.Descriptors
import com.google.protobuf.DynamicMessage

fun mergeJsonWithCustomTimestamp(json: String, builder: DynamicMessage.Builder) {
    // 将JSON解析为结构化JsonObject
    val jsonObject = JsonParser.parseString(json).asJsonObject
    
    val descriptor = builder.descriptorForType
    
    // 遍历所有字段，识别并处理Timestamp类型字段
    descriptor.fields.forEach { field ->
        if (field.type == Descriptors.FieldDescriptor.Type.MESSAGE && 
            field.messageType.fullName == "google.protobuf.Timestamp" &&
            jsonObject.has(field.name)) {
            
            // 将微秒时间戳转换为合法的Timestamp对象
            val micros = jsonObject.get(field.name).asLong
            val timestamp = Timestamps.fromMicros(micros)
            // 手动设置到DynamicMessage Builder
            builder.setField(field, timestamp)
            // 移除该字段，避免后续默认merge重复处理
            jsonObject.remove(field.name)
        }
    }
    
    // 合并剩余非Timestamp字段
    JsonFormat.parser().merge(jsonObject.toString(), builder)
}

修改原代码调用自定义方法 #

val topic = "sample-topic"
val descriptor: Descriptors.Descriptor = getDescriptorFromRegistry(topic)
val builder: DynamicMessage.Builder = DynamicMessage.newBuilder(descriptor)
// 使用自定义方法完成JSON与DynamicMessage的合并
mergeJsonWithCustomTimestamp(json, builder)
val message: DynamicMessage = builder.build()
val bytes: ByteArray = message.toByteArray()

方案说明 #

• 无需修改原始JSON字符串，仅在内存中处理结构化的JSON对象
• 自动识别所有google.protobuf.Timestamp类型的字段，兼容微秒级时间戳转换
• 剩余字段仍使用Protobuf默认的JSON解析逻辑，保证其他类型字段的解析正确性

内容的提问来源于stack exchange，提问作者Yorgos S.