Producing specific avro messages to kafka topic

[theseeker58]

Hello,
eventually I managed to figure out how to marshal and unmarshal structs. But I coudn't find any example that shows how to produce specific records to confluent kafka topics.
I have found this example https://github.com/confluentinc/confluent-kafka-go/blob/master/examples/avro_specific_producer_example/avro_specific_producer_example.go but unfortunately it's based on a different serializer. I like more hamba since it doesn't force the developer to generate code from the schema.
Reading the example I can't see if all I have to do is just assign the Value field of the kafka.Message with the result of the marshal function or somethin more (or different) is needed.
Thank you again for your patience and support

[nrwiersma]

You will not find examples here as it is dependent on your Kafka lib, which is an opinionated topic.
What your are looking for however can be found in this guide: https://medium.com/slalom-technology/introduction-to-schema-registry-in-kafka-915ccf06b902
and the wire format can be found here: https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format

A schema registry client is provided in this package: https://pkg.go.dev/github.com/hamba/avro/v2@v2.16.0/registry so you do not need to implement that, but you may should you choose.
Gos binary package can be used to encode the schema ID: https://pkg.go.dev/encoding/binary#example-ByteOrder-Put

[theseeker58]

Hi,
pretty useful! I managed to produce avro message to a topic after adding magic byte and schema id to the outcome of hamba avro marshalling. The kafka-avro-console-consumer cli actually shows record written to the topic. Now I'm gonna implement in golang the consumer too.
I guess I shoud remove magic byte and schema ID from the received bytes before unmarshalling using hamba, isn't it? Of course first I need to use schema id for dinamically getting schema from the registry
Thanks again

[nrwiersma]

Read the magic byte and check it. Read the schema id, and call the registry GetSchema to get the correct schema, then unmarshal the rest of the payload with the schema you got. The schema registry caches the schema by id (as it is immutable) so it does not download it every time.

[theseeker58]

Thank you. It works perfectly. This is the sample function that the consumer calls

func deserializeInto(payload []byte, emp *Employee) error {
	if payload[0] != 0x0 {
		return errors.New("Unknown magic byte")
	}
	id := binary.BigEndian.Uint32(payload[1:5])
	schema, err := client.GetSchema(context.Background(), int(id))
	if err != nil {
		return err
	}
	value := payload[5:]
	if err := avro.Unmarshal(schema, value, emp); err != nil {
		log.Println("error:", err)
		return err
	}
	return nil
}

As you can see getSchema is invoked each time a new message is consumed from the topic. If I'm not mistaken I don't have to worry about performances since the received schema cached by the library. Is it right?
Now I have 2 sample projects (a producer and a consumer). They use confluent platform (I have the docker-compose yaml file), the confluent kafka client ("github.com/confluentinc/confluent-kafka-go/v2/kafka") but hamba/avro and hamba/registry instead of confluent libraries. In particular gogen-avro is not needed to generate structs. If you think it can be useful as example I can push them in 2 github repos. Let me know...

[nrwiersma]

Nice, I would likely add a length check on payload otherwise a bad message will make the application panic. Something like:

  if len(payload) < 6 {
    return errors.New("message too short")
  }

If I'm not mistaken I don't have to worry about performances since the received schema cached by the library. Is it right?

Yes, this is correct.

If you think it can be useful as example I can push them in 2 github repos. Let me know...

This is up to you. I tend to write these functions as I need them, as often I need to handle other message types as well, like moving from json to avro, etc. As you have discovered, they are very simple to whip up in a couple of minutes 😄

[theseeker58]

Ok, I'm a retired former IT manager. Just studying new stuff for brain work-out :-D. I have time!

[nrwiersma]

Then I am glad you are having fun and learning something new. Always a great hobby 😄