Repository: hamba/avro
Pull request #236 (2023-02-08 21:41:48)

deserialize functionality
Proposing deserialize functionality to client, i.e. the capability to extrapolate the id of the schema from the payload, as described in the [confluent docs regarding the wire format](https://docs.confluent.io/3.2.0/schema-registry/docs/serializer-formatter.html#wire-format).

Basically a kafka message has five "header bytes" in the payload, with the first byte being the magic byte, always zero, and the remaining four representing the id of the schema.

Added a private function

`func extractSchemaIDFromPayload(payload []byte) (int, error)`

which allows to obtain the schema_id as integer, then used by method

`func (c *Client) DeserializePayload(
	ctx context.Context,
	payload []byte,
	target interface{},
) error`

which gets the schema and uses it to deserialize the obtained payload in input.

Thank you for the attention and the work, in case of interest I will be available for the review and the eventual corrections.

Nuc

nrwiersma: Hi,

To start, thanks for the contribution. I have had a good think about this and I am now convinced this does not belong in the `registry.Client`. I would suggest to become its own concept, perhaps `registry.Decoder` that takes a `registry.Client` and optionally an `avro.API`. This allows customisation of how the avro is decoded, but in the simplest case can use the `avro.DefaultConfig`. I think it would also allow for static byte payloads as well as a stream reader mechanics.

What do you think?
nucccc: Thank you for the consideration and the suggestion.
I agree with with what you are saying, while I just implemented a functionality, a separate concept is by far a more reasonable design choice. I suggest the name `Deserializer`, since to what I see that is the name most often used in the Apache Kafka ecosystem, as for example in [this link](https://kafka.apache.org/21/javadoc/org/apache/kafka/common/serialization/Deserializer.html) or [this other link](https://quarkus.io/blog/kafka-serde/).

Given time, I would gladly do my best to implement the concept. Do you think a new file could do the job? Maybe still in the same package `registry`, something like `deserializer.go`.

I'm open to any suggestion or objection! Thank you for the work!
nrwiersma: I would stay away from the name `Deserializer` as Kafka is Java and therefore uses Java naming. In Go `Decoder` is far far more common: see [json.Decoder](https://pkg.go.dev/encoding/json#Decoder) and [go-json.Decoder](https://github.com/goccy/go-json/blob/master/decode.go#L15) and [avro.Decoder](https://github.com/hamba/avro/blob/main/decoder.go#L9).

It is its own concept so a new file `decoder.go` in the `registry` package would be good.
nucccc: Hi,
thank you for the attentions.

I created `decoder.go`, `decoder_test.go` and `decoder_internal_test.go` where I moved everything pertaining to the new `Decoder` struct/concept.

To be initialized it requires a `*registry.Client`, and implements DefaultConfig as the API.
Next, I will try to create a method returning such as 'WithAPI' to return a new decoder with a different API. However I value a lot your opinion on such choice. Up to my limited experience, I would not create two different constructors.

Hoping it to be useful!

Nuc
nucccc: I included the possibility to set the API in the same way as you do for the client, with DecoderFunc options.

I hope the contribution to be useful, for any modification required I'm available, thank you for the consideration.

I saw that there needs to be a merging with the main branch, I don't exactly know who/how is that work supposed to be done, but I'm available for any necessity.

Nuc
nucccc: > Thanks for the contribution, I have left a couple nitpick comments.

I made my best to update everything as requested, and I think to have modified everything accordingly
nucccc: Thank you for the attention and the guidance!

Patch:
diff --git a/registry/decoder.go b/registry/decoder.go
new file mode 100644
index 0000000..43eb87c
--- /dev/null
+++ b/registry/decoder.go
@@ -0,0 +1,79 @@
+package registry
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+
+	"github.com/hamba/avro/v2"
+)
+
+// DecoderFunc is a function used to customize the Decoder.
+type DecoderFunc func(*Decoder)
+
+// WithAPI generates a new decoder with the same client as the one on
+// which the method is invoked and as api the one provided in input.
+func WithAPI(api avro.API) DecoderFunc {
+	return func(d *Decoder) {
+		d.api = api
+	}
+}
+
+// Decoder is entitled to receive raw messages as bytes, obtain the
+// related schema and then use it to deserialize the remaining content.
+type Decoder struct {
+	client *Client
+	api    avro.API
+}
+
+// NewDecoder shall return a new decoder given a client.
+func NewDecoder(client *Client, opts ...DecoderFunc) *Decoder {
+	d := &Decoder{
+		client: client,
+		api:    avro.DefaultConfig,
+	}
+	for _, opt := range opts {
+		opt(d)
+	}
+	return d
+}
+
+// Decode takes in input a payload to be deserialized, extrapolates
+// its schema id, gets the related schema and uses it to unmarshal the payload.
+// The payload shall be formatted according to:
+// https://docs.confluent.io/3.2.0/schema-registry/docs/serializer-formatter.html#wire-format .
+func (d *Decoder) Decode(
+	ctx context.Context,
+	payload []byte,
+	target interface{},
+) error {
+	if len(payload) < 6 {
+		return fmt.Errorf("payload not containing data")
+	}
+
+	id, err := extractSchemaIDFromPayload(payload)
+	if err != nil {
+		return fmt.Errorf("unable to extract schema id from payload, error: %w", err)
+	}
+
+	schema, err := d.client.GetSchema(ctx, id)
+	if err != nil {
+		return fmt.Errorf("unable to obtain schema, error: %w", err)
+	}
+
+	return d.api.Unmarshal(schema, payload[5:], target)
+}
+
+// ExtractSchemaIDFromPayload extrapolates the schema id from a payload composed
+// of raw bytes containing a magic bytes, 4 bytes representing the schema encoding,
+// and the remaining payload being encoded with avro, as described in
+// https://docs.confluent.io/3.2.0/schema-registry/docs/serializer-formatter.html#wire-format .
+func extractSchemaIDFromPayload(payload []byte) (int, error) {
+	if len(payload) < 5 {
+		return 0, fmt.Errorf("payload too short to contain avro header")
+	}
+	if payload[0] != 0 {
+		return 0, fmt.Errorf("magic byte value is %d, different from 0", payload[0])
+	}
+	return int(binary.BigEndian.Uint32(payload[1:5])), nil
+}


Test patch:
diff --git a/registry/decoder_internal_test.go b/registry/decoder_internal_test.go
new file mode 100644
index 0000000..869e7c0
--- /dev/null
+++ b/registry/decoder_internal_test.go
@@ -0,0 +1,42 @@
+package registry
+
+import (
+	"encoding/binary"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+//series of tests for decoding
+
+func TestExtractSchemaIDFromPayload(t *testing.T) {
+	id := 42
+	payload := make([]byte, 5)
+	payload[0] = 0
+
+	binary.BigEndian.PutUint32(payload[1:5], uint32(id))
+
+	extracted_id, err := extractSchemaIDFromPayload(payload)
+
+	require.NoError(t, err)
+
+	assert.Equal(t, extracted_id, id)
+}
+
+func TestExtractSchemaIDFromPayload_HandlesError(t *testing.T) {
+	payload := make([]byte, 4)
+	payload[0] = 0
+
+	_, err := extractSchemaIDFromPayload(payload)
+
+	assert.Equal(t, err.Error(), "payload too short to contain avro header")
+
+	payload = make([]byte, 5)
+	payload[0] = 1
+
+	_, err = extractSchemaIDFromPayload(payload)
+
+	assert.Equal(t, err.Error(), fmt.Sprintf("magic byte value is %d, different from 0", payload[0]))
+}
diff --git a/registry/decoder_test.go b/registry/decoder_test.go
new file mode 100644
index 0000000..4238ecf
--- /dev/null
+++ b/registry/decoder_test.go
@@ -0,0 +1,112 @@
+package registry_test
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/hamba/avro/v2"
+	"github.com/hamba/avro/v2/registry"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDecoder_WithAPI(t *testing.T) {
+	client, err := registry.NewClient("http://example.com")
+	require.NoError(t, err)
+
+	registry.NewDecoder(client, registry.WithAPI(avro.DefaultConfig))
+}
+
+func TestDecoder_Decode(t *testing.T) {
+	//i declare a schema id and a struct to be deserialized
+	schema_id := 42
+	type Person struct {
+		Name string `avro:"name"`
+		Age  int    `avro:"age"`
+	}
+	john := &Person{
+		Name: "john",
+		Age:  28,
+	}
+
+	//i declare the string schema to be returned by the httptest server
+	var person_schema_string string = `{\"type\":\"record\",\"name\":\"person\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"}]}`
+
+	//i declare the string schema to be parsed
+	var person_schema_string_to_parse string = `{"type":"record","name":"person","fields":[{"name":"name","type":"string"},{"name":"age","type":"int"}]}`
+
+	schema, err := avro.Parse(person_schema_string_to_parse)
+
+	//i marshal the payload
+	john_payload, err := avro.Marshal(schema, john)
+	require.NoError(t, err)
+
+	//i then add the magic byte to the payload, together with the encoded id
+	payload := make([]byte, 0, len(john_payload)+5)
+	payload = append(payload, 0)
+	binarySchemaId := make([]byte, 4)
+	binary.BigEndian.PutUint32(binarySchemaId, uint32(schema_id))
+	payload = append(payload, binarySchemaId...)
+	payload = append(payload, john_payload...)
+
+	//instantiating the test server
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "GET", r.Method)
+		assert.Equal(t, fmt.Sprintf("/schemas/ids/%d", schema_id), r.URL.Path)
+
+		//_, _ = w.Write([]byte(fmt.Sprintf(`{"schema":""}`)))
+		_, _ = w.Write([]byte(fmt.Sprintf(`{"schema":"%s"}`, person_schema_string)))
+	}))
+	t.Cleanup(s.Close)
+
+	//creating the test client
+	client, _ := registry.NewClient(s.URL)
+	decoder := registry.NewDecoder(client)
+
+	target_john := &Person{}
+	err = decoder.Decode(context.Background(), payload, target_john)
+
+	require.NoError(t, err)
+	assert.Equal(t, john.Name, target_john.Name)
+	assert.Equal(t, john.Age, target_john.Age)
+}
+
+func TestDecoder_DecodeHandlesError(t *testing.T) {
+	schema_id := 42
+
+	//instantiating the test server
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "GET", r.Method)
+		assert.Equal(t, fmt.Sprintf("/schemas/ids/%d", schema_id), r.URL.Path)
+
+		_, _ = w.Write([]byte(`{"schema":"boh"}`))
+	}))
+	t.Cleanup(s.Close)
+
+	//creating the test client
+	client, _ := registry.NewClient(s.URL)
+	decoder := registry.NewDecoder(client)
+
+	err := decoder.Decode(context.Background(), []byte{0}, nil)
+
+	assert.Equal(t, "payload not containing data", err.Error())
+
+	err = decoder.Decode(context.Background(), []byte{1, 1, 1, 1, 1, 1, 1, 1, 1}, nil)
+
+	assert.Equal(t, "unable to extract schema id from payload, error: magic byte value is 1, different from 0", err.Error())
+
+	payload := make([]byte, 0, 6)
+	payload = append(payload, 0)
+	binarySchemaId := make([]byte, 4)
+	binary.BigEndian.PutUint32(binarySchemaId, uint32(schema_id))
+	payload = append(payload, binarySchemaId...)
+	payload = append(payload, []byte{0}...)
+
+	err = decoder.Decode(context.Background(), payload, nil)
+
+	assert.Equal(t, "unable to obtain schema, error: avro: unknown type: boh", err.Error())
+}