在flink (scala)中使用kryo注册特定类型的protobuf序列化程序时出现问题

import org.apache.flink.api.common.typeutils.TypeSerializer
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.util.serialization.{KeyedDeserializationSchema, KeyedSerializationSchema}
import com.esotericsoftware.kryo.Kryo
import com.twitter.chill.protobuf.ProtobufSerializer

// 定义protobuf对象类型
case class MyProtobufObject(name: String, age: Int)

object FlinkProtobufExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 注册protobuf序列化程序
    val kryo = new Kryo()
    kryo.register(classOf[MyProtobufObject], new ProtobufSerializer[MyProtobufObject](classOf[MyProtobufObject]))

    // 使用KryoSerializer进行序列化
    val serializer: TypeSerializer[MyProtobufObject] = new KryoSerializer[MyProtobufObject](classOf[MyProtobufObject], kryo)

    // 使用注册的序列化程序进行序列化和反序列化
    val dataStream = env.socketTextStream("localhost", 9999)
      .map(value => {
        // 将数据转换为protobuf对象
        val protobufObject = MyProtobufObject.newBuilder()
          .setName(value.split(",")(0))
          .setAge(value.split(",")(1).toInt)
          .build()
        protobufObject
      })
      .map(value => {
        // 序列化为字节数组
        serializer.serialize(value)
      })
      .map(value => {
        // 反序列化为protobuf对象
        serializer.deserialize(value)
      })

    // 执行任务
    env.execute("Flink Protobuf Example")
  }
}