Migrate to hamba/avro (#346)

Author: mostafa

avro.go

@@ -1,5 +1,11 @@
 package kafka
 
+import (
+	"encoding/json"
+
+	"github.com/hamba/avro/v2"
+)
+
 type AvroSerde struct {
 	Serdes
 }
@@ -11,12 +17,19 @@ func (*AvroSerde) Serialize(data interface{}, schema *Schema) ([]byte, *Xk6Kafka
 		return nil, err
 	}
 
-	encodedData, _, originalErr := schema.Codec().NativeFromTextual(jsonBytes)
-	if originalErr != nil {
-		return nil, NewXk6KafkaError(failedToEncode, "Failed to encode data", originalErr)
+	avroSchema := schema.Codec()
+	if avroSchema == nil {
+		return nil, NewXk6KafkaError(failedToEncode, "Failed to parse Avro schema", nil)
 	}
 
-	bytesData, originalErr := schema.Codec().BinaryFromNative(nil, encodedData)
+	// Parse JSON data into a map for marshaling
+	var jsonData interface{}
+	if jsonErr := json.Unmarshal(jsonBytes, &jsonData); jsonErr != nil {
+		return nil, NewXk6KafkaError(failedToEncode, "Failed to parse JSON data", jsonErr)
+	}
+
+	// Marshal to binary using hamba/avro
+	bytesData, originalErr := avro.Marshal(avroSchema, jsonData)
 	if originalErr != nil {
 		return nil, NewXk6KafkaError(failedToEncodeToBinary,
 			"Failed to encode data into binary",
@@ -28,7 +41,13 @@ func (*AvroSerde) Serialize(data interface{}, schema *Schema) ([]byte, *Xk6Kafka
 
 // Deserialize deserializes a Avro binary into a JSON object.
 func (*AvroSerde) Deserialize(data []byte, schema *Schema) (interface{}, *Xk6KafkaError) {
-	decodedData, _, err := schema.Codec().NativeFromBinary(data)
+	avroSchema := schema.Codec()
+	if avroSchema == nil {
+		return nil, NewXk6KafkaError(failedToDecodeFromBinary, "Failed to parse Avro schema", nil)
+	}
+
+	var decodedData interface{}
+	err := avro.Unmarshal(avroSchema, data, &decodedData)
 	if err != nil {
 		return nil, NewXk6KafkaError(
 			failedToDecodeFromBinary, "Failed to decode data", err)
@@ -37,6 +56,6 @@ func (*AvroSerde) Deserialize(data []byte, schema *Schema) (interface{}, *Xk6Kaf
 	if data, ok := decodedData.(map[string]interface{}); ok {
 		return data, nil
 	} else {
-		return nil, ErrInvalidDataType
+		return decodedData, nil
 	}
 }

avro_test.go

@@ -62,8 +62,8 @@ func TestSerializeAvroFailsOnValidation(t *testing.T) {
 	actual, err := avroSerde.Serialize(map[string]interface{}{"value": "key"}, schema)
 	assert.Nil(t, actual)
 	assert.NotNil(t, err)
-	assert.Equal(t, "Failed to encode data", err.Message)
-	assert.Equal(t, failedToEncode, err.Code)
+	assert.Equal(t, "Failed to encode data into binary", err.Message)
+	assert.Equal(t, failedToEncodeToBinary, err.Code)
 }
 
 // TestDeserializeAvro tests the deserialization of a JSON object from Avro binary.

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.39.0
 	github.com/aws/aws-sdk-go-v2/config v1.31.7
 	github.com/grafana/sobek v0.0.0-20250723111835-dd8a13f0d439
-	github.com/linkedin/goavro/v2 v2.14.0
+	github.com/hamba/avro/v2 v2.29.0
 	github.com/pavlo-v-chernykh/keystore-go/v4 v4.5.0
 	github.com/riferrei/srclient v0.7.3
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
@@ -41,15 +41,20 @@ require (
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
 	github.com/golang/snappy v1.0.0 // indirect
 	github.com/google/pprof v0.0.0-20230728192033-2ba5b33183c6 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/linkedin/goavro/v2 v2.13.1 // indirect
 	github.com/mailru/easyjson v0.9.0 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mstoykov/atlas v0.0.0-20220811071828-388f114305dd // indirect
 	github.com/mstoykov/k6-taskqueue-lib v0.1.3 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect

go.sum

@@ -7,7 +7,7 @@ import (
 	"net/http"
 
 	"github.com/grafana/sobek"
-	"github.com/linkedin/goavro/v2"
+	"github.com/hamba/avro/v2"
 	"github.com/riferrei/srclient"
 	"github.com/santhosh-tekuri/jsonschema/v5"
 	"go.k6.io/k6/js/common"
@@ -50,7 +50,7 @@ type Schema struct {
 	Version       int                  `json:"version"`
 	References    []srclient.Reference `json:"references"`
 	Subject       string               `json:"subject"`
-	codec         *goavro.Codec
+	avroSchema    avro.Schema
 	jsonSchema    *jsonschema.Schema
 }
 
@@ -66,17 +66,17 @@ type WireFormat struct {
 	Data     []byte `json:"data"`
 }
 
-// Codec ensures access to Codec
+// Codec ensures access to parsed Avro Schema
 // Will try to initialize a new one if it hasn't been initialized before
-// Will return nil if it can't initialize a codec from the schema
-func (s *Schema) Codec() *goavro.Codec {
-	if s.codec == nil {
-		codec, err := goavro.NewCodec(s.Schema)
+// Will return nil if it can't initialize a schema from the schema string
+func (s *Schema) Codec() avro.Schema {
+	if s.avroSchema == nil {
+		schema, err := avro.Parse(s.Schema)
 		if err == nil {
-			s.codec = codec
+			s.avroSchema = schema
 		}
 	}
-	return s.codec
+	return s.avroSchema
 }
 
 // JsonSchema ensures access to JsonSchema

schema_registry.go

@@ -207,7 +207,7 @@ func TestSerializeFails(t *testing.T) {
 				},
 				SchemaType: srclient.Avro,
 			},
-			err: NewXk6KafkaError(failedToEncode, "Failed to encode data", errors.New("cannot decode textual record \"io.confluent.kafka.avro.Schema\": cannot decode textual map: cannot determine codec: \"unknown\"")),
+			err: NewXk6KafkaError(failedToEncodeToBinary, "Failed to encode data into binary", errors.New("avro: missing required field key")),
 		},
 		{
 			container: &Container{