feat: use hamba encoder schema function if available (#1440)

Author: agouil

schemaregistry/serde/avrov2/avro.go

@@ -105,13 +105,17 @@ func (s *Serializer) SerializeWithHeaders(topic string, msg interface{}) ([]kafk
 	if s.Conf.UseSchemaID == -1 &&
 		!s.Conf.UseLatestVersion &&
 		len(s.Conf.UseLatestWithMetadata) == 0 {
-		msgType := reflect.TypeOf(msg)
-		if msgType.Kind() != reflect.Pointer {
-			return nil, nil, errors.New("input message must be a pointer")
-		}
-		avroSchema, err = StructToSchema(msgType.Elem())
-		if err != nil {
-			return nil, nil, err
+		if st, ok := msg.(interface{ Schema() avro.Schema }); ok {
+			avroSchema = st.Schema()
+		} else {
+			msgType := reflect.TypeOf(msg)
+			if msgType.Kind() != reflect.Pointer {
+				return nil, nil, errors.New("input message must be a pointer")
+			}
+			avroSchema, err = StructToSchema(msgType.Elem())
+			if err != nil {
+				return nil, nil, err
+			}
 		}
 		info = schemaregistry.SchemaInfo{
 			Schema: avroSchema.String(),

schemaregistry/serde/avrov2/avro_test.go

@@ -18,11 +18,13 @@ package avrov2
 
 import (
 	"errors"
-	"github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/rules/cel"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/rules/cel"
+	"github.com/hamba/avro/v2"
+
 	_ "github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/rules/cel"
 	"github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/rules/encryption"
 	_ "github.com/confluentinc/confluent-kafka-go/v2/schemaregistry/rules/encryption/awskms"
@@ -286,7 +288,62 @@ const (
       "confluent:tags": [ "PII" ]
     }
   ]
-} 
+}
+`
+	demoWithSchemaFuncSchema = `
+{
+  "name": "DemoWithSchemaFunc",
+  "type": "record",
+  "fields": [
+    {
+      "name": "IntField",
+      "type": "int"
+    },
+    {
+      "name": "BoolField",
+      "type": "boolean"
+    },
+    {
+      "name": "ArrayField",
+      "type": {
+        "type": "array",
+        "items": "string"
+      }
+    },
+    {
+      "name": "MapField",
+      "type": {
+        "type": "map",
+        "values": "string"
+      }
+    },
+    {
+      "name": "StringField",
+      "type": ["null", "string"]
+    },
+    {
+      "name": "EnumField",
+      "type": {
+        "name": "GreetingsEnum",
+        "type": "enum",
+        "symbols": ["hey", "bye"]
+      }
+    },
+    {
+      "name": "RecordField",
+      "type": {
+        "name": "GreetingsObj",
+        "type": "record",
+        "fields": [
+          {
+            "name": "Hey",
+            "type": "string"
+          }
+        ]
+      }
+    }
+  ]
+}
 `
 )
 
@@ -302,6 +359,8 @@ func testMessageFactory(subject string, name string) (interface{}, error) {
 		return &DemoSchemaSingleTag{}, nil
 	case "DemoSchemaWithUnion":
 		return &DemoSchemaWithUnion{}, nil
+	case "DemoWithSchemaFunc":
+		return &DemoWithSchemaFunc{}, nil
 	case "ComplexSchema":
 		return &ComplexSchema{}, nil
 	case "SchemaEvolution":
@@ -804,6 +863,46 @@ func TestAvroSchemaEvolution(t *testing.T) {
 	serde.MaybeFail("deserialization", err, serde.Expect(msg, &obj2))
 }
 
+func TestAvroSerdeWithEncodingSchemaFunc(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)
+
+	ser, err := NewSerializer(client, serde.ValueSerde, NewSerializerConfig())
+	serde.MaybeFail("Serializer configuration", err)
+
+	obj := DemoWithSchemaFunc{
+		IntField:    123,
+		StringField: nil,
+		BoolField:   true,
+		ArrayField:  []string{"hello", "world"},
+		MapField: map[string]string{
+			"hello": "world",
+		},
+		EnumField: "hey",
+		RecordField: struct {
+			Hey string `json:"Hey"`
+		}{Hey: "bye"},
+	}
+	bytes, err := ser.Serialize("topic1", &obj)
+	serde.MaybeFail("serialization", err)
+
+	deser, err := NewDeserializer(client, serde.ValueSerde, NewDeserializerConfig())
+	serde.MaybeFail("Deserializer configuration", err)
+	deser.Client = ser.Client
+	deser.MessageFactory = testMessageFactory
+
+	var newobj DemoWithSchemaFunc
+	err = deser.DeserializeInto("topic1", bytes, &newobj)
+	serde.MaybeFail("deserialization into", err, serde.Expect(newobj, obj))
+
+	msg, err := deser.Deserialize("topic1", bytes)
+	serde.MaybeFail("deserialization", err, serde.Expect(msg, &obj))
+}
+
 func TestAvroSerdeWithCELCondition(t *testing.T) {
 	serde.MaybeFail = serde.InitFailFunc(t)
 	var err error
@@ -2688,3 +2787,19 @@ type SchemaEvolution1 struct {
 type SchemaEvolution2 struct {
 	NewOptionalField string `json:"NewOptionalField"`
 }
+
+type DemoWithSchemaFunc struct {
+	IntField    int32             `json:"IntField"`
+	BoolField   bool              `json:"BoolField"`
+	StringField *string           `json:"StringField"`
+	ArrayField  []string          `json:"ArrayField"`
+	MapField    map[string]string `json:"MapField"`
+	EnumField   string            `json:"EnumField"`
+	RecordField struct {
+		Hey string `json:"Hey"`
+	} `json:"RecordField"`
+}
+
+func (d *DemoWithSchemaFunc) Schema() avro.Schema {
+	return avro.MustParse(demoWithSchemaFuncSchema)
+}