No need for Schema Registry in your Spring-Kafka tests
Spring framework has great support for testing your Spring application with Apache Kafka. You add spring-kafka-dependency in your maven pom.xml file and you annotate your test class with @EmbbededKafka and Spring will do the rest.
Spring will start embedded Kafka on some random port and you are ready to test your application. It is not a real production-ready Kafka, but for testing, it is enough. You can test if your application is sending data to the correct topic. You can test whether your application has received data from the topic.
But if your application uses Apache Avro for serialization, and you do not use Confluent Schema Registry (which is the case when you want to use spring-kafka-test for testing) then you have a problem. There is no Serializer/Deserializer provided by the Spring framework for such a use case. Even if you just want to test your application, your application needs to have access to Confluent Schema Registry to work.
Problematic SchemaRegistryClient
A common setup for your Spring application with Avro and Kafka is the following. You search on the Internet what Serializer/Deserializer you need to use if you want to use Avro with Kafka. You find that you need to use io.confluent.kafka.serializers.KafkaAvroSerializer and io.confluent.kafka.serializers.KafkaAvroDeserializer. So you put those 2 in your application.properties file.
Then you try to test your Spring application (Kafka + Avro) against embedded Kafka and you encounter the following exception:
KafkaException: failed to construct kafka consumer
Well, you forgot to define “schema.registry.url” in your properties file.
So you put some imaginary URL hoping it will work. It doesn’t work. You will get the following exception.
org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
Let’s save some time and jump straight to the problem. The problem is in the CashedSchemaRegistryClient class. It is used by the KafkaAvroSerializer/KafkaAvroDeserializer to fetch schema definition from the Confluent Schema Registry. We can not control which SchemaRegistryClient is used in our application. But we can control which Serializer/Deserialized will be used by our application.
And there is the solution. The solution is to create your own KafkaAvroSerializer/KafkaAvroDeserializer. One that will know how to serialize/deserialize the data without going for the schema definition to Schema Registry. We already have schemas we use locally. There is no need to go for schemas to some external source.
Custom KafkaAvroSerializer
Creating CustomKafkaAvroSerializer is easy. You extend existing KafkaAvroSerializer and you tell him to use
io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.
For this to work, we need to do one more step. You need to tell your Spring application (application.properties) to use created CustomKafkaAvroSerializer. We will do that later. Let’s first create our own KafkaAvroDeserializer.
Custom KafkaAvroDeserializer
Creating CustomKafkaAvroDeserialized is a bit more complicated. Your application can read from many topics. If you have many topics, then you have many Avro Schemas to choose from. Your application needs to know which Avro Schema to use when deserializing data. And our Deserializer will do just that. It will map the Avro Schema to the topic. If the data comes from topic-1 we will use one schema. If the data comes from topic-2 we will use another one.
In code, it looks like this. We will extend the KafkaAvroDeserializer and overwrite Object deserialize(String topic, byte[] bytes) method. In the deserialize method, KafkaAvroDeserializer would go for schema definition to Schema Registry. This is not needed. We know which Avro schema to use from the topic name. We return that information by supplying MockSchemaRegistryClient with the selected schema. And that is it. Our KafkaAvroDeserializer will know how to deserialize the data.
Use created class in your test context
The last step is to update your application.properties file with the created classes.
And voilà, you are now able to test your Spring Kafka application using Avro Serialization against embedded Kafka provided by Spring.
