handle empty payload (#1145)

Author: fzmoment

schemaregistry/serde/protobuf/protobuf.go

@@ -357,7 +357,7 @@ func NewDeserializer(client schemaregistry.Client, serdeType serde.Type, conf *D
 
 // Deserialize implements deserialization of Protobuf data
 func (s *Deserializer) Deserialize(topic string, payload []byte) (interface{}, error) {
-	if payload == nil {
+	if len(payload) == 0 {
 		return nil, nil
 	}
 	info, err := s.GetSchema(topic, payload)

schemaregistry/serde/protobuf/protobuf_test.go

@@ -184,3 +184,18 @@ func TestProtobufSerdeWithCycle(t *testing.T) {
 	newobj, err := deser.Deserialize("topic1", bytes)
 	serde.MaybeFail("deserialization", err, serde.Expect(newobj.(proto.Message).ProtoReflect(), obj.ProtoReflect()))
 }
+
+func TestProtobufSerdeEmptyMessage(t *testing.T) {
+	serde.MaybeFail = serde.InitFailFunc(t)
+	var err error
+	conf := schemaregistry.NewConfig("mock://")
+	client, err := schemaregistry.NewClient(conf)
+	serde.MaybeFail("Schema Registry configuration", err)
+	deser, err := NewDeserializer(client, serde.ValueSerde, NewDeserializerConfig())
+	serde.MaybeFail("Deserializer configuration", err)
+
+	_, err = deser.Deserialize("topic1", nil)
+	serde.MaybeFail("deserialization", err)
+	_, err = deser.Deserialize("topic1", []byte{})
+	serde.MaybeFail("deserialization", err)
+}