Kafka Schema Registry

[src: conduktor.io]

Using Avro schema with Kafka producers and consumers is a common practice to ensure data serialization and deserialization compatibility across different applications. Avro is a data serialization format that allows you to define data schemas in a compact and efficient way.

Here are the steps to use Avro schema with Kafka producer and consumer:

1. Define the Avro Schema: First, you need to define the Avro schema for your data. Avro schemas are typically defined using JSON or the Avro schema definition language (.avsc). The schema describes the structure of the data you want to send through Kafka.

Example Avro schema for a simple user record:

{
  "type": "record",
  "name": "User",
  "fields": [
    { "name": "id", "type": "int" },
    { "name": "name", "type": "string" },
    { "name": "age", "type": "int" }
  ]
}

To generate Avro classes from Avro schema using the Java plugin, you can follow these steps:

1. Add the Avro Maven plugin to your project’s pom.xml file:

<build>
  <plugins>
    <plugin>
      <groupId>org.apache.avro</groupId>
      <artifactId>avro-maven-plugin</artifactId>
      <version>1.10.2</version>
      <executions>
        <execution>
          <phase>generate-sources</phase>
          <goals>
            <goal>schema</goal>
            <goal>protocol</goal>
            <goal>idl-protocol</goal>
            <goal>idl</goal>
          </goals>
        </execution>
      </executions>
      <configuration>
        <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
        <outputDirectory>${project.build.directory}/generated-sources</outputDirectory>
        <stringType>String</stringType> <!-- Optional: Specify the string type mapping -->
        <enableDecimalLogicalType>true</enableDecimalLogicalType> <!-- Optional: Enable support for decimal logical type -->
      </configuration>
    </plugin>
  </plugins>
</build>

2. Create a directory called avro inside your project's src/main directory. Place your Avro schema file(s) inside this directory with .avsc extension.

3. Run the Maven build command to generate the Avro classes:

mvn clean generate-sources

This will execute the Avro Maven plugin and generate the Avro classes in the specified output directory (${project.build.directory}/generated-sources).

4. After the build is successful, you can find the generated Avro classes in the output directory. These classes can be used to work with Avro data in your Java code.

Note: Make sure you have Maven installed and configured on your system before running the Maven build command.

3. Set up Kafka Producer: In the Kafka Producer application, you need to configure the producer to use Avro serialization. For this, you’ll need to include Avro-related libraries like Avro and Kafka Avro Serializer in your project.

Example Java code to set up a Kafka producer using Avro serialization:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

Producer<String, User> producer = new KafkaProducer<>(props);

1. Serialize and Send Avro Data: When producing messages, you can use the Avro classes to create Avro data and send it as the message value.

Example Java code to produce Avro data:

User user = new User(1, "John Doe", 30);
ProducerRecord<String, User> record = new ProducerRecord<>("topic-name", user.getId().toString(), user);
producer.send(record);

4. Set up Kafka Consumer: Similarly, on the consumer side, you need to configure the consumer to use Avro deserialization.

Example Java code to set up a Kafka consumer using Avro deserialization:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("schema.registry.url", "http://localhost:8081");
props.put("specific.avro.reader", "true");

KafkaConsumer<String, User> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("topic-name"));

1. Deserialize and Consume Avro Data: When consuming messages, the Kafka Avro deserializer will handle the deserialization and provide you with Avro data.

Example Java code to consume Avro data:

while (true) {
    ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, User> record : records) {
        User user = record.value();
        System.out.println("Received user: " + user.getName() + ", Age: " + user.getAge());
    }
}

References

What is the Schema Registry and why do you need to use it?

Level Up Coding

Thanks for being a part of our community! Before you go:

• 👏 Clap for the story and follow the author 👉
• 📰 View more content in the Level Up Coding publication
• 💰 Free coding interview course ⇒ View Course
• 🧠 AI Tools ⇒ View Now
• 🔔 Follow us: Twitter | LinkedIn | Newsletter

🚀👉 Join the Level Up talent collective and find an amazing job