java serialization apache-kafka avro spring-kafka

When serializing a Kafka Avro object I then get "org.apache.kafka.common.errors.SerializationException: Unknown magic byte!"

I have a Java Spring application that uses Kafka to consume Avro messages.

In my tests I serialize an Avro object to a byte[] and send it with a KafkaTemplate<String, byte[]> to the topic.

This is how I serialize the object:

private static byte[] serialize(Product product) {
    var specificDatumWriter = new SpecificDatumWriter<>(Product.class);
    try (var byteArrayOutputStream = new ByteArrayOutputStream()) {
        var binaryEncoder = EncoderFactory.get().directBinaryEncoder(byteArrayOutputStream, null);
        specificDatumWriter.write(product, binaryEncoder);
        binaryEncoder.flush();
        return byteArrayOutputStream.toByteArray();
    } catch (IOException e) {
        throw new IllegalStateException(e);
    }
}

kafkaTemplate.send("products", uuid.toString(), serialize(product));

I have followed this example for serializing the object: https://www.baeldung.com/java-apache-avro#1-serialization

However, I get this error message when my @KafkaListener tries to deserialize the message:

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener failed; nested exception is org.springframework.kafka.support.serializer.DeserializationException: failed to deserialize; nested exception is org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:2060)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:2045)
    ...
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.kafka.support.serializer.DeserializationException: failed to deserialize; nested exception is org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
    at org.springframework.kafka.support.serializer.ErrorHandlingDeserializer.deserializationException(ErrorHandlingDeserializer.java:216)
    at org.springframework.kafka.support.serializer.ErrorHandlingDeserializer.deserialize(ErrorHandlingDeserializer.java:191)
    ...
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1232)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1127)
    ... 4 more
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

What am I missing? What am I doing wrong with the serialization?

Solution

I have now got it working. I have to use the KafkaAvroSerializer to serialize the object.

private byte[] serialize(Product product) {
    try (var kafkaAvroSerializer = new KafkaAvroSerializer()) {
        Map<String, Object> props = new HashMap<>();
        props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
        props.put(KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS, false);
        props.put(KafkaAvroSerializerConfig.VALUE_SUBJECT_NAME_STRATEGY, RecordNameStrategy.class);
        kafkaAvroSerializer.configure(props, false);
        return kafkaAvroSerializer.serialize("products", product);
    }
}