[FLINK-34954] Kryo input implementation NoFetchingInput fails to handle zero length bytes

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.19.0
• Fix Version/s: 1.20.0
• Component/s: API / Type Serialization System
• Labels: pull-request-available
• Assignee: Qinghui Xu
• Reporter: Qinghui Xu
• Created: 27/Mar/24 19:17
• Updated: 21/May/24 08:42
• Resolved: 19/Apr/24 09:56

Description

If the serailized bytes are empty, `NoFetchingInput` will run into error when Kryo tries to deserialize it.

Example: a protobuf 3 object that contains only default values will be serialized as 0 length byte array, and the deserialization later will fail. Illustration:

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.{ByteBufferInput, ByteBufferOutput, Input, Output}
import com.google.protobuf.{DescriptorProtos, Message}import com.twitter.chill.protobuf.ProtobufSerializer
import org.apache.flink.api.java.typeutils.runtime.NoFetchingInput
import java.io.ByteArrayInputStream
 
object ProtoSerializationTest {
  def main(args: Array[String]) = {     
    val chillProtoSerializer = new ProtobufSerializer
    val protomessage = DescriptorProtos.DescriptorProto.getDefaultInstance
    val output: Output = new ByteBufferOutput(1000)
    chillProtoSerializer.write(null, output, protomessage)
    val serialized: Array[Byte] = output.toBytes
    println(s"Serialized : $serialized")
    val input: Input = new NoFetchingInput(new ByteArrayInputStream(serialized))
    val deserialized = chillProtoSerializer.read(null, input, classOf[BillableClick].asInstanceOf[Class[Message]])
    println(deserialized)
  }
}

 

Error

Exception in thread "main" java.lang.RuntimeException: Could not create class com.criteo.glup.BillableClickProto$BillableClick
    at com.twitter.chill.protobuf.ProtobufSerializer.read(ProtobufSerializer.java:76)
    at com.criteo.streaming.common.bootstrap.ProtoSerialization$.main(ProtoSerialization.scala:22)
    at ProtoSerialization.main(ProtoSerialization.scala)
Caused by: com.esotericsoftware.kryo.KryoException: java.io.EOFException: No more bytes left.
    at org.apache.flink.api.java.typeutils.runtime.NoFetchingInput.readBytes(NoFetchingInput.java:128)
    at com.esotericsoftware.kryo.io.Input.readBytes(Input.java:332)
    at com.twitter.chill.protobuf.ProtobufSerializer.read(ProtobufSerializer.java:73)
    ... 2 more
Caused by: java.io.EOFException: No more bytes left.
    ... 5 more

Issue Links

• causes: FLINK-35215 The performance of serializerKryo and serializerKryoWithoutRegistration are regressed
• links to: GitHub Pull Request #24586